/*-------------------------------------------------------------------------
 *
 * dbcommands.c
 *		数据库管理命令（创建/删除数据库）。
 *
 * 注意：数据库创建/销毁命令对数据库对象使用独占锁（由 LockSharedObject() 表达）以避免互相干扰。  以前我们对 pg_database 使用表级锁，但这太粗糙了。
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  src/backend/commands/dbcommands.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include <fcntl.h>
#include <unistd.h>
#include <sys/stat.h>

#include "access/genam.h"
#include "access/heapam.h"
#include "access/htup_details.h"
#include "access/multixact.h"
#include "access/tableam.h"
#include "access/xact.h"
#include "access/xloginsert.h"
#include "access/xlogrecovery.h"
#include "access/xlogutils.h"
#include "catalog/catalog.h"
#include "catalog/dependency.h"
#include "catalog/indexing.h"
#include "catalog/objectaccess.h"
#include "catalog/pg_authid.h"
#include "catalog/pg_collation.h"
#include "catalog/pg_database.h"
#include "catalog/pg_db_role_setting.h"
#include "catalog/pg_subscription.h"
#include "catalog/pg_tablespace.h"
#include "commands/comment.h"
#include "commands/dbcommands.h"
#include "commands/dbcommands_xlog.h"
#include "commands/defrem.h"
#include "commands/seclabel.h"
#include "commands/tablespace.h"
#include "common/file_perm.h"
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "postmaster/bgwriter.h"
#include "replication/slot.h"
#include "storage/copydir.h"
#include "storage/fd.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "storage/md.h"
#include "storage/procarray.h"
#include "storage/smgr.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
#include "utils/pg_locale.h"
#include "utils/relmapper.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"

/*
 * 创建数据库策略。
 *
 * CREATEDB_WAL_LOG将在块级别复制数据库，并对每个
 * 复制的块进行WAL日志记录。
 *
 * CREATEDB_FILE_COPY将简单地执行文件系统级别的复制
 * 数据库，并对每个复制的表空间记录一条记录。为了确保安全，
 * 它还会在操作之前和之后触发检查点。
 */
typedef enum CreateDBStrategy
{
	CREATEDB_WAL_LOG,
	CREATEDB_FILE_COPY
} CreateDBStrategy;

typedef struct
{
	Oid			src_dboid;		/* 源（模板）数据库 */
	Oid			dest_dboid;		/* 我们尝试创建的数据库 */
	CreateDBStrategy strategy;	/* 创建数据库策略 */
} createdb_failure_params;

typedef struct
{
	Oid			dest_dboid;		/* 我们尝试移动的数据库 */
	Oid			dest_tsoid;		/* 我们尝试移动到的表空间 */
} movedb_failure_params;

/*
 * 创建数据库时要复制的关系的信息。
 */
typedef struct CreateDBRelInfo
{
	RelFileNode rnode;			/* 物理关系标识符 */
	Oid			reloid;			/* 关系oid */
	bool		permanent;		/* 关系是永久的还是非日志的 */
} CreateDBRelInfo;


/* 非导出函数原型 */
static void fc_createdb_failure_callback(int fc_code, Datum fc_arg);
static void fc_movedb(const char *fc_dbname, const char *fc_tblspcname);
static void fc_movedb_failure_callback(int fc_code, Datum fc_arg);
static bool fc_get_db_info(const char *fc_name, LOCKMODE fc_lockmode,
						Oid *fc_dbIdP, Oid *fc_ownerIdP,
						int *fc_encodingP, bool *fc_dbIsTemplateP, bool *fc_dbAllowConnP,
						TransactionId *fc_dbFrozenXidP, MultiXactId *fc_dbMinMultiP,
						Oid *fc_dbTablespace, char **fc_dbCollate, char **fc_dbCtype, char **fc_dbIculocale,
						char *fc_dbLocProvider,
						char **fc_dbCollversion);
static bool fc_have_createdb_privilege(void);
static void fc_remove_dbtablespaces(Oid fc_db_id);
static bool fc_check_db_file_conflict(Oid fc_db_id);
static int	fc_errdetail_busy_db(int fc_notherbackends, int fc_npreparedxacts);
static void fc_CreateDatabaseUsingWalLog(Oid fc_src_dboid, Oid fc_dboid, Oid fc_src_tsid,
									  Oid fc_dst_tsid);
static List *fc_ScanSourceDatabasePgClass(Oid fc_srctbid, Oid fc_srcdbid, char *fc_srcpath);
static List *fc_ScanSourceDatabasePgClassPage(Page fc_page, Buffer fc_buf, Oid fc_tbid,
										   Oid fc_dbid, char *fc_srcpath,
										   List *fc_rnodelist, Snapshot fc_snapshot);
static CreateDBRelInfo *fc_ScanSourceDatabasePgClassTuple(HeapTupleData *fc_tuple,
													   Oid fc_tbid, Oid fc_dbid,
													   char *fc_srcpath);
static void fc_CreateDirAndVersionFile(char *fc_dbpath, Oid fc_dbid, Oid fc_tsid,
									bool fc_isRedo);
static void fc_CreateDatabaseUsingFileCopy(Oid fc_src_dboid, Oid fc_dboid, Oid fc_src_tsid,
										Oid fc_dst_tsid);
static void fc_recovery_create_dbdir(char *fc_path, bool fc_only_tblspc);

/*
 * 使用WAL_LOG策略创建一个新数据库。
 *
 * 每个复制的块都会单独写入预写日志。
 */
static void fc_CreateDatabaseUsingWalLog(Oid fc_src_dboid, Oid fc_dst_dboid,
						  Oid fc_src_tsid, Oid fc_dst_tsid)
{
	char	   *fc_srcpath;
	char	   *fc_dstpath;
	List	   *fc_rnodelist = NULL;
	ListCell   *fc_cell;
	LockRelId	fc_srcrelid;
	LockRelId	fc_dstrelid;
	RelFileNode fc_srcrnode;
	RelFileNode fc_dstrnode;
	CreateDBRelInfo *fc_relinfo;

	/* 获取源数据库和目标数据库路径。 */
	fc_srcpath = GetDatabasePath(fc_src_dboid, fc_src_tsid);
	fc_dstpath = GetDatabasePath(fc_dst_dboid, fc_dst_tsid);

	/* 创建数据库目录并写入PG_VERSION文件。 */
	fc_CreateDirAndVersionFile(fc_dstpath, fc_dst_dboid, fc_dst_tsid, false);

	/* 从源数据库复制relmap文件到目标数据库。 */
	RelationMapCopy(fc_dst_dboid, fc_dst_tsid, fc_srcpath, fc_dstpath);

	/* 获取要从源数据库复制的relfilenodes列表。 */
	fc_rnodelist = fc_ScanSourceDatabasePgClass(fc_src_tsid, fc_src_dboid, fc_srcpath);
	Assert(fc_rnodelist != NIL);

	/*
	 * 数据库ID对于所有关系来说是相同的，因此在进入循环之前
	 * 需要先设置它们。
	 */
	fc_srcrelid.dbId = fc_src_dboid;
	fc_dstrelid.dbId = fc_dst_dboid;

	/* 遍历我们的relfilenodes列表并复制每个。 */
	foreach(fc_cell, fc_rnodelist)
	{
		fc_relinfo = lfirst(fc_cell);
		fc_srcrnode = fc_relinfo->rnode;

		/*
		 * 如果关系来自源数据库的默认表空间，则我们
		 * 需要在目标数据库的默认表空间中创建它。
		 * 否则，我们需要在与源数据库相同的表空间中创建。
		 */
		if (fc_srcrnode.spcNode == fc_src_tsid)
			fc_dstrnode.spcNode = fc_dst_tsid;
		else
			fc_dstrnode.spcNode = fc_srcrnode.spcNode;

		fc_dstrnode.dbNode = fc_dst_dboid;
		fc_dstrnode.relNode = fc_srcrnode.relNode;

		/*
		 * 在复制之前获取源和目标关系的锁。
		 *
		 * 我们通常不会在没有
		 * 持有关系锁的情况下将关系数据读取到shared_buffers中。
		 * 如果在这种情况下跳过它可能会出什么问题并不明确，
		 * 因为此时没有人可以修改源数据库或目标数据库，
		 * 而且我们对两个数据库都有锁，但我们还是采取保守的做法。
		 */
		fc_dstrelid.relId = fc_srcrelid.relId = fc_relinfo->reloid;
		LockRelationId(&fc_srcrelid, AccessShareLock);
		LockRelationId(&fc_dstrelid, AccessShareLock);

		/* 从源复制关系存储到目标。 */
		CreateAndCopyRelationData(fc_srcrnode, fc_dstrnode, fc_relinfo->permanent);

		/* 释放关系锁。 */
		UnlockRelationId(&fc_srcrelid, AccessShareLock);
		UnlockRelationId(&fc_dstrelid, AccessShareLock);
	}

	pfree(fc_srcpath);
	pfree(fc_dstpath);
	list_free_deep(fc_rnodelist);
}

/*
 * 扫描源数据库的 pg_class 表以识别需要复制到目标数据库的关系。
 *
 * 这是对通常不可跨数据库访问规则的一个例外。我们之所以能让它在这里工作，是因为我们知道没有连接到源数据库，并且（由于不存在准备事务触及该数据库）也没有不确定的元组。这意味着我们不需要担心修剪从我们下面删除任何东西，并且我们对快照的要求也不需要过于苛刻。只要它将所有之前提交的 XIDs 视为已提交，将所有中止的 XIDs 视为已中止，我们就没问题：这里不可能有其他情况。
 *
 * 我们不能在这里依赖 relcache，因为它只知道我们连接到的数据库，并且无法处理对其他数据库的访问。这也意味着我们不能依赖堆扫描基础设施，反正这也是个坏主意，因为它可能尝试执行诸如 HOT 修剪之类的操作，而我们在没有连接的数据库中是绝对无法安全执行的。
 */
static List * fc_ScanSourceDatabasePgClass(Oid fc_tbid, Oid fc_dbid, char *fc_srcpath)
{
	RelFileNode fc_rnode;
	BlockNumber fc_nblocks;
	BlockNumber fc_blkno;
	Buffer		fc_buf;
	Oid			fc_relfilenode;
	Page		fc_page;
	List	   *fc_rnodelist = NIL;
	LockRelId	fc_relid;
	Snapshot	fc_snapshot;
	SMgrRelation	fc_smgr;
	BufferAccessStrategy fc_bstrategy;

	/* 获取 pg_class relfilenode。 */
	fc_relfilenode = RelationMapOidToFilenodeForDatabase(fc_srcpath,
													  RelationRelationId);

	/* 在没有持有关系锁的情况下不要将数据读入 shared_buffers。 */
	fc_relid.dbId = fc_dbid;
	fc_relid.relId = RelationRelationId;
	LockRelationId(&fc_relid, AccessShareLock);

	/* 为 pg_class 关系准备一个 RelFileNode。 */
	fc_rnode.spcNode = fc_tbid;
	fc_rnode.dbNode = fc_dbid;
	fc_rnode.relNode = fc_relfilenode;

	fc_smgr = smgropen(fc_rnode, InvalidBackendId);
	fc_nblocks = smgrnblocks(fc_smgr, MAIN_FORKNUM);
	smgrclose(fc_smgr);

	/* 由于这是一次批量读取操作，因此使用缓冲访问策略。 */
	fc_bstrategy = GetAccessStrategy(BAS_BULKREAD);

	/*
	 * 正如函数头注释中所解释的，我们需要一个快照，它将看到所有提交的事务视为已提交，我们的事务快照 - 或活动快照 - 可能不够新，但 GetLatestSnapshot() 的返回值应该可以正常工作。
	 */
	fc_snapshot = GetLatestSnapshot();

	/* 按块处理关系。 */
	for (fc_blkno = 0; fc_blkno < fc_nblocks; fc_blkno++)
	{
		CHECK_FOR_INTERRUPTS();

		fc_buf = ReadBufferWithoutRelcache(fc_rnode, MAIN_FORKNUM, fc_blkno,
										RBM_NORMAL, fc_bstrategy, true);

		LockBuffer(fc_buf, BUFFER_LOCK_SHARE);
		fc_page = BufferGetPage(fc_buf);
		if (PageIsNew(fc_page) || PageIsEmpty(fc_page))
		{
			UnlockReleaseBuffer(fc_buf);
			continue;
		}

		/* 将当前页面的相关 pg_class 元组附加到 rnodelist。 */
		fc_rnodelist = fc_ScanSourceDatabasePgClassPage(fc_page, fc_buf, fc_tbid, fc_dbid,
												  fc_srcpath, fc_rnodelist,
												  fc_snapshot);

		UnlockReleaseBuffer(fc_buf);
	}

	/* 释放关系锁。 */
	UnlockRelationId(&fc_relid, AccessShareLock);

	return fc_rnodelist;
}

/*
 * 扫描源数据库的 pg_class 关系的一页，并将相关条目添加到 rnodelist。返回值是更新后的列表。
 */
static List * fc_ScanSourceDatabasePgClassPage(Page fc_page, Buffer fc_buf, Oid fc_tbid, Oid fc_dbid,
							  char *fc_srcpath, List *fc_rnodelist,
							  Snapshot fc_snapshot)
{
	BlockNumber fc_blkno = BufferGetBlockNumber(fc_buf);
	OffsetNumber fc_offnum;
	OffsetNumber fc_maxoff;
	HeapTupleData fc_tuple;

	fc_maxoff = PageGetMaxOffsetNumber(fc_page);

	/* 循环偏移量。 */
	for (fc_offnum = FirstOffsetNumber;
		 fc_offnum <= fc_maxoff;
		 fc_offnum = OffsetNumberNext(fc_offnum))
	{
		ItemId		fc_itemid;

		fc_itemid = PageGetItemId(fc_page, fc_offnum);

		/* 如果槽位为空或已经死亡，则无事可做。 */
		if (!ItemIdIsUsed(fc_itemid) || ItemIdIsDead(fc_itemid) ||
			ItemIdIsRedirected(fc_itemid))
			continue;

		Assert(ItemIdIsNormal(fc_itemid));
		ItemPointerSet(&(fc_tuple.t_self), fc_blkno, fc_offnum);

		/* 初始化一个 HeapTupleData 结构。 */
		fc_tuple.t_data = (HeapTupleHeader) PageGetItem(fc_page, fc_itemid);
		fc_tuple.t_len = ItemIdGetLength(fc_itemid);
		fc_tuple.t_tableOid = RelationRelationId;

		/* 跳过此快照不可见的元组。 */
		if (HeapTupleSatisfiesVisibility(&fc_tuple, fc_snapshot, fc_buf))
		{
			CreateDBRelInfo *fc_relinfo;

			/*
			 * ScanSourceDatabasePgClassTuple 负责为此元组构建
			 * CreateDBRelInfo 对象，但也可以决定这个元组不是我们需要复制的。如果我们确实需要复制关系，将其添加到列表中。
			 */
			fc_relinfo = fc_ScanSourceDatabasePgClassTuple(&fc_tuple, fc_tbid, fc_dbid,
													 fc_srcpath);
			if (fc_relinfo != NULL)
				fc_rnodelist = lappend(fc_rnodelist, fc_relinfo);
		}
	}

	return fc_rnodelist;
}

/*
 * 决定某个 pg_class 元组是否表示需要从源数据库复制到目标数据库的内容，如果是，则为其构建一个 CreateDBRelInfo。
 *
 * 可见性检查由调用者负责，因此我们在这里的工作只是评估元组中存储的数据。
 */
CreateDBRelInfo *
fc_ScanSourceDatabasePgClassTuple(HeapTupleData *fc_tuple, Oid fc_tbid, Oid fc_dbid,
							   char *fc_srcpath)
{
	CreateDBRelInfo *fc_relinfo;
	Form_pg_class fc_classForm;
	Oid			fc_relfilenode = InvalidOid;

	fc_classForm = (Form_pg_class) GETSTRUCT(fc_tuple);

	/*
	 * 如果这个对象不需要被复制则返回 NULL。
	 *
	 * 共享对象不需要被复制，因为它们是共享的。
	 * 没有存储的对象无法被复制，因为没有任何东西可以复制。
	 * 临时关系也不需要被复制，因为它们在创建它们的会话之外不可访问，那个会话必须已经终止，并且如果它仍然存在，无法连接到不同的数据库。 autovacuum 最终也会移除 pg_class 条目。
	 */
	if (fc_classForm->reltablespace == GLOBALTABLESPACE_OID ||
		!RELKIND_HAS_STORAGE(fc_classForm->relkind) ||
		fc_classForm->relpersistence == RELPERSISTENCE_TEMP)
		return NULL;

	/*
	 * 如果 relfilenode 是有效的，则直接使用它。否则，查询 relmap。
	 */
	if (OidIsValid(fc_classForm->relfilenode))
		fc_relfilenode = fc_classForm->relfilenode;
	else
		fc_relfilenode = RelationMapOidToFilenodeForDatabase(fc_srcpath,
														  fc_classForm->oid);

	/* 我们必须有一个有效的 relfilenode oid。 */
	if (!OidIsValid(fc_relfilenode))
		elog(ERROR, "relation with OID %u does not have a valid relfilenode",
			 fc_classForm->oid);

	/* 准备一个关系信息元素并将其添加到列表中。 */
	fc_relinfo = (CreateDBRelInfo *) palloc(sizeof(CreateDBRelInfo));
	if (OidIsValid(fc_classForm->reltablespace))
		fc_relinfo->rnode.spcNode = fc_classForm->reltablespace;
	else
		fc_relinfo->rnode.spcNode = fc_tbid;

	fc_relinfo->rnode.dbNode = fc_dbid;
	fc_relinfo->rnode.relNode = fc_relfilenode;
	fc_relinfo->reloid = fc_classForm->oid;

	/* 临时关系在上面被拒绝。 */
	Assert(fc_classForm->relpersistence != RELPERSISTENCE_TEMP);
	fc_relinfo->permanent =
		(fc_classForm->relpersistence == RELPERSISTENCE_PERMANENT) ? true : false;

	return fc_relinfo;
}

/*
 * 创建数据库目录并在数据库路径中写入 PG_VERSION 文件。如果 isRedo 为真，则数据库目录已存在也没关系。
 */
static void fc_CreateDirAndVersionFile(char *fc_dbpath, Oid fc_dbid, Oid fc_tsid, bool fc_isRedo)
{
	int			fc_fd;
	int			fc_nbytes;
	char		fc_versionfile[MAXPGPATH];
	char		fc_buf[16];

	/*
	 * 请注意，我们不必从源数据库复制版本数据；只有一个合法值。
	 */
	sprintf(fc_buf, "%s\n", PG_MAJORVERSION);
	fc_nbytes = strlen(PG_MAJORVERSION) + 1;

	/* 创建数据库目录。 */
	if (MakePGDirectory(fc_dbpath) < 0)
	{
		/* 除了已经存在或不在 WAL 重放中的失败？ */
		if (errno != EEXIST || !fc_isRedo)
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not create directory \"%s\": %m", fc_dbpath)));
	}

	/*
	 * 在数据库路径中创建 PG_VERSION 文件。如果文件已存在并且我们在 WAL 重放中，则再尝试以写模式打开它。
	 */
	snprintf(fc_versionfile, sizeof(fc_versionfile), "%s/%s", fc_dbpath, "PG_VERSION");

	fc_fd = OpenTransientFile(fc_versionfile, O_WRONLY | O_CREAT | O_EXCL | PG_BINARY);
	if (fc_fd < 0 && errno == EEXIST && fc_isRedo)
		fc_fd = OpenTransientFile(fc_versionfile, O_WRONLY | O_TRUNC | PG_BINARY);

	if (fc_fd < 0)
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not create file \"%s\": %m", fc_versionfile)));

	/* 在 PG_VERSION 文件中写入 PG_MAJORVERSION。 */
	pgstat_report_wait_start(WAIT_EVENT_VERSION_FILE_WRITE);
	errno = 0;
	if ((int) write(fc_fd, fc_buf, fc_nbytes) != fc_nbytes)
	{
		/* 如果写入未设置 errno，假设问题是没有磁盘空间。 */
		if (errno == 0)
			errno = ENOSPC;
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not write to file \"%s\": %m", fc_versionfile)));
	}
	pgstat_report_wait_end();

	pgstat_report_wait_start(WAIT_EVENT_VERSION_FILE_SYNC);
	if (pg_fsync(fc_fd) != 0)
		ereport(data_sync_elevel(ERROR),
				(errcode_for_file_access(),
				 errmsg("could not fsync file \"%s\": %m", fc_versionfile)));
	fsync_fname(fc_dbpath, true);
	pgstat_report_wait_end();

	/* 关闭版本文件。 */
	CloseTransientFile(fc_fd);

	/* 如果我们不在 WAL 重放中，则写入 WAL。 */
	if (!fc_isRedo)
	{
		xl_dbase_create_wal_log_rec fc_xlrec;

		START_CRIT_SECTION();

		fc_xlrec.db_id = fc_dbid;
		fc_xlrec.tablespace_id = fc_tsid;

		XLogBeginInsert();
		XLogRegisterData((char *) (&fc_xlrec),
						 sizeof(xl_dbase_create_wal_log_rec));

		(void) XLogInsert(RM_DBASE_ID, XLOG_DBASE_CREATE_WAL_LOG);

		END_CRIT_SECTION();
	}
}

/*
 * 使用 FILE_COPY 策略创建一个新数据库。
 *
 * 在文件系统级别复制每个表空间，并为每个复制的表空间记录一个 WAL 记录。这需要在复制之前和之后进行检查点，这可能是昂贵的，但如果复制的数据库很大，它确实大大减少了 WAL 生成。
 */
static void fc_CreateDatabaseUsingFileCopy(Oid fc_src_dboid, Oid fc_dst_dboid, Oid fc_src_tsid,
							Oid fc_dst_tsid)
{
	TableScanDesc fc_scan;
	Relation	fc_rel;
	HeapTuple	fc_tuple;

	/*
	 * 在开始复制之前强制执行检查点。这将强制所有脏缓冲区，包括未登录表的缓冲区，写入磁盘，以确保源数据库在磁盘上的内容是最新的以供复制。
	 * FlushDatabaseBuffers() 可以满足这一点，但我们还想处理任何挂起的 unlink 请求。否则，如果在我们复制文件时发生检查点，文件可能会在我们即将复制它时被删除，导致 copydir() 中的 lstat() 调用因 ENOENT 失败。
	 */
	RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE |
					  CHECKPOINT_WAIT | CHECKPOINT_FLUSH_ALL);

	/*
	 * 遍历模板数据库的所有表空间，并将每个表空间复制到新数据库中。
	 */
	fc_rel = table_open(TableSpaceRelationId, AccessShareLock);
	fc_scan = table_beginscan_catalog(fc_rel, 0, NULL);
	while ((fc_tuple = heap_getnext(fc_scan, ForwardScanDirection)) != NULL)
	{
		Form_pg_tablespace fc_spaceform = (Form_pg_tablespace) GETSTRUCT(fc_tuple);
		Oid			fc_srctablespace = fc_spaceform->oid;
		Oid			fc_dsttablespace;
		char	   *fc_srcpath;
		char	   *fc_dstpath;
		struct stat fc_st;

		/* 无需复制全局表空间 */
		if (fc_srctablespace == GLOBALTABLESPACE_OID)
			continue;

		fc_srcpath = GetDatabasePath(fc_src_dboid, fc_srctablespace);

		if (stat(fc_srcpath, &fc_st) < 0 || !S_ISDIR(fc_st.st_mode) ||
			directory_is_empty(fc_srcpath))
		{
			/* 假设我们可以忽略它 */
			pfree(fc_srcpath);
			continue;
		}

		if (fc_srctablespace == fc_src_tsid)
			fc_dsttablespace = fc_dst_tsid;
		else
			fc_dsttablespace = fc_srctablespace;

		fc_dstpath = GetDatabasePath(fc_dst_dboid, fc_dsttablespace);

		/*
		 * 将此子目录复制到新位置
		 *
		 * 我们不需要复制子目录
		 */
		copydir(fc_srcpath, fc_dstpath, false);

		/* 在 XLOG 中记录文件系统更改 */
		{
			xl_dbase_create_file_copy_rec fc_xlrec;

			fc_xlrec.db_id = fc_dst_dboid;
			fc_xlrec.tablespace_id = fc_dsttablespace;
			fc_xlrec.src_db_id = fc_src_dboid;
			fc_xlrec.src_tablespace_id = fc_srctablespace;

			XLogBeginInsert();
			XLogRegisterData((char *) &fc_xlrec,
							 sizeof(xl_dbase_create_file_copy_rec));

			(void) XLogInsert(RM_DBASE_ID,
							  XLOG_DBASE_CREATE_FILE_COPY | XLR_SPECIAL_REL_UPDATE);
		}
		pfree(fc_srcpath);
		pfree(fc_dstpath);
	}
	table_endscan(fc_scan);
	table_close(fc_rel, AccessShareLock);

	/*
	 * 我们在提交之前强制进行检查点。这实际上意味着
	 * 已提交的 XLOG_DBASE_CREATE_FILE_COPY 操作将不需要被
	 * 重放（至少在普通的崩溃恢复中不需要；我们仍然需要
	 * 为了 PITR 操作而进行 XLOG 记录）。这避免了
	 * 两种麻烦的情况：
	 *
	 * #1: 当 PITR 关闭时，我们不记录新创建的
	 * 索引的内容；因此 DBASE_CREATE 重放的删除并重新创建整个目录的行为会丢失这些索引。
	 *
	 * #2: 由于我们必须在 DBASE_CREATE
	 * 重放期间重新复制源数据库，我们有风险复制在原始 CREATE DATABASE 命令后但在导致重放的系统崩溃之前已提交的更改。
	 * 这至少是不预期的，最糟糕的情况可能导致不一致，例如重复的表名。
	 *
	 * （这两者在 8.0 到 8.0.3 的版本中都是实际的 bug）
	 *
	 * 在 PITR 重放中，第一种情况不是问题，第二种情况只有在
	 * CREATE DATABASE 和随后的模板数据库
	 * 更改都发生在基础备份正在进行时才存在风险。
	 * 除了将其记录为一种限制外，我们似乎无能为力。
	 *
	 * 请参阅 CreateDatabaseUsingWalLog()，以获取一种
	 * 避免这些问题的更优雅的 CREATE DATABASE 策略。
	 */
	RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT);
}

/*
 * CREATE DATABASE
 */
Oid createdb(ParseState *fc_pstate, const CreatedbStmt *fc_stmt)
{
	Oid			fc_src_dboid;
	Oid			fc_src_owner;
	int			fc_src_encoding = -1;
	char	   *fc_src_collate = NULL;
	char	   *fc_src_ctype = NULL;
	char	   *fc_src_iculocale = NULL;
	char		fc_src_locprovider = '\0';
	char	   *fc_src_collversion = NULL;
	bool		fc_src_istemplate;
	bool		fc_src_allowconn;
	TransactionId fc_src_frozenxid = InvalidTransactionId;
	MultiXactId fc_src_minmxid = InvalidMultiXactId;
	Oid			fc_src_deftablespace;
	volatile Oid fc_dst_deftablespace;
	Relation	fc_pg_database_rel;
	HeapTuple	fc_tuple;
	Datum		fc_new_record[Natts_pg_database];
	bool		fc_new_record_nulls[Natts_pg_database];
	Oid			fc_dboid = InvalidOid;
	Oid			fc_datdba;
	ListCell   *fc_option;
	DefElem    *fc_dtablespacename = NULL;
	DefElem    *fc_downer = NULL;
	DefElem    *fc_dtemplate = NULL;
	DefElem    *fc_dencoding = NULL;
	DefElem    *fc_dlocale = NULL;
	DefElem    *fc_dcollate = NULL;
	DefElem    *fc_dctype = NULL;
	DefElem    *fc_diculocale = NULL;
	DefElem    *fc_dlocprovider = NULL;
	DefElem    *fc_distemplate = NULL;
	DefElem    *fc_dallowconnections = NULL;
	DefElem    *fc_dconnlimit = NULL;
	DefElem    *fc_dcollversion = NULL;
	DefElem    *fc_dstrategy = NULL;
	char	   *fc_dbname = fc_stmt->dbname;
	char	   *fc_dbowner = NULL;
	const char *fc_dbtemplate = NULL;
	char	   *fc_dbcollate = NULL;
	char	   *fc_dbctype = NULL;
	char	   *fc_dbiculocale = NULL;
	char		fc_dblocprovider = '\0';
	char	   *fc_canonname;
	int			fc_encoding = -1;
	bool		fc_dbistemplate = false;
	bool		fc_dballowconnections = true;
	int			fc_dbconnlimit = DATCONNLIMIT_UNLIMITED;
	char	   *fc_dbcollversion = NULL;
	int			fc_notherbackends;
	int			fc_npreparedxacts;
	CreateDBStrategy fc_dbstrategy = CREATEDB_WAL_LOG;
	createdb_failure_params fc_fparms;

	/* 从语句节点树中提取选项 */
	foreach(fc_option, fc_stmt->options)
	{
		DefElem    *fc_defel = (DefElem *) lfirst(fc_option);

		if (strcmp(fc_defel->defname, "tablespace") == 0)
		{
			if (fc_dtablespacename)
				errorConflictingDefElem(fc_defel, fc_pstate);
			fc_dtablespacename = fc_defel;
		}
		else if (strcmp(fc_defel->defname, "owner") == 0)
		{
			if (fc_downer)
				errorConflictingDefElem(fc_defel, fc_pstate);
			fc_downer = fc_defel;
		}
		else if (strcmp(fc_defel->defname, "template") == 0)
		{
			if (fc_dtemplate)
				errorConflictingDefElem(fc_defel, fc_pstate);
			fc_dtemplate = fc_defel;
		}
		else if (strcmp(fc_defel->defname, "encoding") == 0)
		{
			if (fc_dencoding)
				errorConflictingDefElem(fc_defel, fc_pstate);
			fc_dencoding = fc_defel;
		}
		else if (strcmp(fc_defel->defname, "locale") == 0)
		{
			if (fc_dlocale)
				errorConflictingDefElem(fc_defel, fc_pstate);
			fc_dlocale = fc_defel;
		}
		else if (strcmp(fc_defel->defname, "lc_collate") == 0)
		{
			if (fc_dcollate)
				errorConflictingDefElem(fc_defel, fc_pstate);
			fc_dcollate = fc_defel;
		}
		else if (strcmp(fc_defel->defname, "lc_ctype") == 0)
		{
			if (fc_dctype)
				errorConflictingDefElem(fc_defel, fc_pstate);
			fc_dctype = fc_defel;
		}
		else if (strcmp(fc_defel->defname, "icu_locale") == 0)
		{
			if (fc_diculocale)
				errorConflictingDefElem(fc_defel, fc_pstate);
			fc_diculocale = fc_defel;
		}
		else if (strcmp(fc_defel->defname, "locale_provider") == 0)
		{
			if (fc_dlocprovider)
				errorConflictingDefElem(fc_defel, fc_pstate);
			fc_dlocprovider = fc_defel;
		}
		else if (strcmp(fc_defel->defname, "is_template") == 0)
		{
			if (fc_distemplate)
				errorConflictingDefElem(fc_defel, fc_pstate);
			fc_distemplate = fc_defel;
		}
		else if (strcmp(fc_defel->defname, "allow_connections") == 0)
		{
			if (fc_dallowconnections)
				errorConflictingDefElem(fc_defel, fc_pstate);
			fc_dallowconnections = fc_defel;
		}
		else if (strcmp(fc_defel->defname, "connection_limit") == 0)
		{
			if (fc_dconnlimit)
				errorConflictingDefElem(fc_defel, fc_pstate);
			fc_dconnlimit = fc_defel;
		}
		else if (strcmp(fc_defel->defname, "collation_version") == 0)
		{
			if (fc_dcollversion)
				errorConflictingDefElem(fc_defel, fc_pstate);
			fc_dcollversion = fc_defel;
		}
		else if (strcmp(fc_defel->defname, "location") == 0)
		{
			ereport(WARNING,
					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
					 errmsg("LOCATION is not supported anymore"),
					 errhint("Consider using tablespaces instead."),
					 parser_errposition(fc_pstate, fc_defel->location)));
		}
		else if (strcmp(fc_defel->defname, "oid") == 0)
		{
			fc_dboid = defGetObjectId(fc_defel);

			/*
			 * 我们通常不允许使用
			 * 系统分配的 OID 创建新数据库。pg_upgrade 尝试保留数据库
			 * OID，因此我们不能允许任何数据库以可能在
			 * 将来版本创建的全新初始化集群中使用的 OID 创建。
			 * 我们假设所有这样的 OID 都将来自
			 * 系统管理的 OID 范围。
			 *
			 * 然而，作为例外，当 allow_system_table_mods=on 时（以便 initdb 可以将系统
			 * OID 分配给 template0 和 postgres）或在执行二进制
			 * 升级时（以便 pg_upgrade 可以保留它在源集群中找到的任何 OID）允许分配任何 OID。
			 */
			if (fc_dboid < FirstNormalObjectId &&
				!allowSystemTableMods && !IsBinaryUpgrade)
				ereport(ERROR,
						(errcode(ERRCODE_INVALID_PARAMETER_VALUE)),
						errmsg("OIDs less than %u are reserved for system objects", FirstNormalObjectId));
		}
		else if (strcmp(fc_defel->defname, "strategy") == 0)
		{
			if (fc_dstrategy)
				errorConflictingDefElem(fc_defel, fc_pstate);
			fc_dstrategy = fc_defel;
		}
		else
			ereport(ERROR,
					(errcode(ERRCODE_SYNTAX_ERROR),
					 errmsg("option \"%s\" not recognized", fc_defel->defname),
					 parser_errposition(fc_pstate, fc_defel->location)));
	}

	if (fc_downer && fc_downer->arg)
		fc_dbowner = defGetString(fc_downer);
	if (fc_dtemplate && fc_dtemplate->arg)
		fc_dbtemplate = defGetString(fc_dtemplate);
	if (fc_dencoding && fc_dencoding->arg)
	{
		const char *fc_encoding_name;

		if (IsA(fc_dencoding->arg, Integer))
		{
			fc_encoding = defGetInt32(fc_dencoding);
			fc_encoding_name = pg_encoding_to_char(fc_encoding);
			if (strcmp(fc_encoding_name, "") == 0 ||
				pg_valid_server_encoding(fc_encoding_name) < 0)
				ereport(ERROR,
						(errcode(ERRCODE_UNDEFINED_OBJECT),
						 errmsg("%d is not a valid encoding code",
								fc_encoding),
						 parser_errposition(fc_pstate, fc_dencoding->location)));
		}
		else
		{
			fc_encoding_name = defGetString(fc_dencoding);
			fc_encoding = pg_valid_server_encoding(fc_encoding_name);
			if (fc_encoding < 0)
				ereport(ERROR,
						(errcode(ERRCODE_UNDEFINED_OBJECT),
						 errmsg("%s is not a valid encoding name",
								fc_encoding_name),
						 parser_errposition(fc_pstate, fc_dencoding->location)));
		}
	}
	if (fc_dlocale && fc_dlocale->arg)
	{
		fc_dbcollate = defGetString(fc_dlocale);
		fc_dbctype = defGetString(fc_dlocale);
	}
	if (fc_dcollate && fc_dcollate->arg)
		fc_dbcollate = defGetString(fc_dcollate);
	if (fc_dctype && fc_dctype->arg)
		fc_dbctype = defGetString(fc_dctype);
	if (fc_diculocale && fc_diculocale->arg)
		fc_dbiculocale = defGetString(fc_diculocale);
	if (fc_dlocprovider && fc_dlocprovider->arg)
	{
		char	   *fc_locproviderstr = defGetString(fc_dlocprovider);

		if (pg_strcasecmp(fc_locproviderstr, "icu") == 0)
			fc_dblocprovider = COLLPROVIDER_ICU;
		else if (pg_strcasecmp(fc_locproviderstr, "libc") == 0)
			fc_dblocprovider = COLLPROVIDER_LIBC;
		else
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
					 errmsg("unrecognized locale provider: %s",
							fc_locproviderstr)));
	}
	if (fc_distemplate && fc_distemplate->arg)
		fc_dbistemplate = defGetBoolean(fc_distemplate);
	if (fc_dallowconnections && fc_dallowconnections->arg)
		fc_dballowconnections = defGetBoolean(fc_dallowconnections);
	if (fc_dconnlimit && fc_dconnlimit->arg)
	{
		fc_dbconnlimit = defGetInt32(fc_dconnlimit);
		if (fc_dbconnlimit < DATCONNLIMIT_UNLIMITED)
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("invalid connection limit: %d", fc_dbconnlimit)));
	}
	if (fc_dcollversion)
		fc_dbcollversion = defGetString(fc_dcollversion);

	/* 获取拟议所有者的 OID */
	if (fc_dbowner)
		fc_datdba = get_role_oid(fc_dbowner, false);
	else
		fc_datdba = GetUserId();

	/*
	 * 要创建数据库，必须具有 createdb 权限并且必须能够
	 * 成为目标角色（这并不意味着目标角色本身
	 * 必须具有 createdb 权限）。后者的规定防范了
	 * “赠送”攻击。请注意，超级用户将自然而然地始终具备这两种权限。
	 */
	if (!fc_have_createdb_privilege())
		ereport(ERROR,
				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
				 errmsg("permission denied to create database")));

	check_is_member_of_role(GetUserId(), fc_datdba);

	/*
	 * 查找要克隆的数据库（模板），并获得对其的共享锁。
	 * ShareLock 允许两个 CREATE DATABASE 同时使用同一个模板
	 * 的同时，确保没有人正在并行删除它
	 * （这将是非常糟糕的，因为我们可能会得到一个不完整的副本
	 * 而不知道）。这也防止在我们完成复制之前建立任何新的连接，
	 * 所以我们可以确保它不会在我们之下改变。
	 */
	if (!fc_dbtemplate)
		fc_dbtemplate = "template1";	/* 默认模板数据库名称 */

	if (!fc_get_db_info(fc_dbtemplate, ShareLock,
					 &fc_src_dboid, &fc_src_owner, &fc_src_encoding,
					 &fc_src_istemplate, &fc_src_allowconn,
					 &fc_src_frozenxid, &fc_src_minmxid, &fc_src_deftablespace,
					 &fc_src_collate, &fc_src_ctype, &fc_src_iculocale, &fc_src_locprovider,
					 &fc_src_collversion))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_DATABASE),
				 errmsg("template database \"%s\" does not exist",
						fc_dbtemplate)));

	/*
	 * 如果源数据库正在被删除，我们不能将其用作模板。
	 */
	if (database_is_invalid_oid(fc_src_dboid))
		ereport(ERROR,
				errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				errmsg("cannot use invalid database \"%s\" as template", fc_dbtemplate),
				errhint("Use DROP DATABASE to drop invalid databases."));

	/*
	 * 权限检查：要复制未标记为 datistemplate 的数据库，
	 * 您必须是超级用户或其所有者。
	 */
	if (!fc_src_istemplate)
	{
		if (!pg_database_ownercheck(fc_src_dboid, GetUserId()))
			ereport(ERROR,
					(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
					 errmsg("permission denied to copy database \"%s\"",
							fc_dbtemplate)));
	}

	/* 验证数据库创建策略。 */
	if (fc_dstrategy && fc_dstrategy->arg)
	{
		char	   *fc_strategy;

		fc_strategy = defGetString(fc_dstrategy);
		if (pg_strcasecmp(fc_strategy, "wal_log") == 0)
			fc_dbstrategy = CREATEDB_WAL_LOG;
		else if (pg_strcasecmp(fc_strategy, "file_copy") == 0)
			fc_dbstrategy = CREATEDB_FILE_COPY;
		else
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("invalid create database strategy \"%s\"", fc_strategy),
					 errhint("Valid strategies are \"wal_log\" and \"file_copy\".")));
	}

	/* 如果编码或区域设置使用默认值，请使用源的设置 */
	if (fc_encoding < 0)
		fc_encoding = fc_src_encoding;
	if (fc_dbcollate == NULL)
		fc_dbcollate = fc_src_collate;
	if (fc_dbctype == NULL)
		fc_dbctype = fc_src_ctype;
	if (fc_dblocprovider == '\0')
		fc_dblocprovider = fc_src_locprovider;
	if (fc_dbiculocale == NULL && fc_dblocprovider == COLLPROVIDER_ICU)
		fc_dbiculocale = fc_src_iculocale;

	/* 一些编码仅限于客户端 */
	if (!PG_VALID_BE_ENCODING(fc_encoding))
		ereport(ERROR,
				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
				 errmsg("invalid server encoding %d", fc_encoding)));

	/* 检查所选的区域设置是否有效，并获取标准拼写 */
	if (!check_locale(LC_COLLATE, fc_dbcollate, &fc_canonname))
		ereport(ERROR,
				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
				 errmsg("invalid locale name: \"%s\"", fc_dbcollate)));
	fc_dbcollate = fc_canonname;
	if (!check_locale(LC_CTYPE, fc_dbctype, &fc_canonname))
		ereport(ERROR,
				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
				 errmsg("invalid locale name: \"%s\"", fc_dbctype)));
	fc_dbctype = fc_canonname;

	check_encoding_locale_matches(fc_encoding, fc_dbcollate, fc_dbctype);

	if (fc_dblocprovider == COLLPROVIDER_ICU)
	{
		if (!(is_encoding_supported_by_icu(fc_encoding)))
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("encoding \"%s\" is not supported with ICU provider",
							pg_encoding_to_char(fc_encoding))));

		/*
		 * 如果 template0 使用 libc 提供程序，但新
		 * 数据库使用 icu，则会发生这种情况。
		 */
		if (!fc_dbiculocale)
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("ICU locale must be specified")));

		check_icu_locale(fc_dbiculocale);
	}
	else
	{
		if (fc_dbiculocale)
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
					 errmsg("ICU locale cannot be specified unless locale provider is ICU")));
	}

	/*
	 * 检查新的编码和区域设置是否与源
	 * 数据库匹配。我们坚持这样做是因为我们只是复制源数据 ---
	 * 任何非 ASCII 数据都将被错误编码，并且任何根据源区域设置排序的索引都将是错误的。
	 *
	 * 但是，我们假设 template0 不包含任何非 ASCII 数据
	 * 也不包含依赖于排序或字形的索引，因此 template0 可以
	 * 用作创建任何编码或区域设置的数据库的模板。
	 */
	if (strcmp(fc_dbtemplate, "template0") != 0)
	{
		if (fc_encoding != fc_src_encoding)
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("new encoding (%s) is incompatible with the encoding of the template database (%s)",
							pg_encoding_to_char(fc_encoding),
							pg_encoding_to_char(fc_src_encoding)),
					 errhint("Use the same encoding as in the template database, or use template0 as template.")));

		if (strcmp(fc_dbcollate, fc_src_collate) != 0)
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("new collation (%s) is incompatible with the collation of the template database (%s)",
							fc_dbcollate, fc_src_collate),
					 errhint("Use the same collation as in the template database, or use template0 as template.")));

		if (strcmp(fc_dbctype, fc_src_ctype) != 0)
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("new LC_CTYPE (%s) is incompatible with the LC_CTYPE of the template database (%s)",
							fc_dbctype, fc_src_ctype),
					 errhint("Use the same LC_CTYPE as in the template database, or use template0 as template.")));

		if (fc_dblocprovider != fc_src_locprovider)
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("new locale provider (%s) does not match locale provider of the template database (%s)",
							collprovider_name(fc_dblocprovider), collprovider_name(fc_src_locprovider)),
					 errhint("Use the same locale provider as in the template database, or use template0 as template.")));

		if (fc_dblocprovider == COLLPROVIDER_ICU)
		{
			Assert(fc_dbiculocale);
			Assert(fc_src_iculocale);
			if (strcmp(fc_dbiculocale, fc_src_iculocale) != 0)
				ereport(ERROR,
						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
						 errmsg("new ICU locale (%s) is incompatible with the ICU locale of the template database (%s)",
								fc_dbiculocale, fc_src_iculocale),
						 errhint("Use the same ICU locale as in the template database, or use template0 as template.")));
		}
	}

	/*
	 * 如果我们得到了模板数据库的排序版本，请检查它是否
	 * 与实际的操作系统排序版本匹配。否则报错；用户
	 * 需要先修复模板数据库。如果排序版本被明确指定为语句选项，请不要抱怨；
	 * 这被 pg_upgrade 用来精确复制旧状态。
	 *
	 * （如果模板数据库没有排序版本，那么要么是
	 * 平台/提供程序不支持排序版本，要么是
	 * template0，我们规定它不包含
	 * 使用排序的对象。）
	 */
	if (fc_src_collversion && !fc_dcollversion)
	{
		char	   *fc_actual_versionstr;

		fc_actual_versionstr = get_collation_actual_version(fc_dblocprovider, fc_dblocprovider == COLLPROVIDER_ICU ? fc_dbiculocale : fc_dbcollate);
		if (!fc_actual_versionstr)
#ifdef FDD //cppcheck
		{
#endif
			ereport(ERROR,
					(errmsg("template database \"%s\" has a collation version, but no actual collation version could be determined",
							fc_dbtemplate)));
#ifdef FDD //cppcheck
			return InvalidOid;
		}
#endif

		if (strcmp(fc_actual_versionstr, fc_src_collversion) != 0)
			ereport(ERROR,
					(errmsg("template database \"%s\" has a collation version mismatch",
							fc_dbtemplate),
					 errdetail("The template database was created using collation version %s, "
							   "but the operating system provides version %s.",
							   fc_src_collversion, fc_actual_versionstr),
					 errhint("Rebuild all objects in the template database that use the default collation and run "
							 "ALTER DATABASE %s REFRESH COLLATION VERSION, "
							 "or build PostgreSQL with the right library version.",
							 quote_identifier(fc_dbtemplate))));
	}

	if (fc_dbcollversion == NULL)
		fc_dbcollversion = fc_src_collversion;

	/*
	 * 通常，我们从模板数据库复制排序版本。
	 * 这最后的手段仅适用于模板数据库没有
	 * 排序版本的情况，通常只能在 template0 的情况下出现。
	 */
	if (fc_dbcollversion == NULL)
		fc_dbcollversion = get_collation_actual_version(fc_dblocprovider, fc_dblocprovider == COLLPROVIDER_ICU ? fc_dbiculocale : fc_dbcollate);

	/* 为新数据库解析默认表空间 */
	if (fc_dtablespacename && fc_dtablespacename->arg)
	{
		char	   *fc_tablespacename;
		AclResult	fc_aclresult;

		fc_tablespacename = defGetString(fc_dtablespacename);
		fc_dst_deftablespace = get_tablespace_oid(fc_tablespacename, false);
		/* 检查权限 */
		fc_aclresult = pg_tablespace_aclcheck(fc_dst_deftablespace, GetUserId(),
										   ACL_CREATE);
		if (fc_aclresult != ACLCHECK_OK)
			aclcheck_error(fc_aclresult, OBJECT_TABLESPACE,
						   fc_tablespacename);

		/* pg_global 永远不能是默认表空间 */
		if (fc_dst_deftablespace == GLOBALTABLESPACE_OID)
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("pg_global cannot be used as default tablespace")));

		/*
		 * 如果我们尝试更改模板的默认表空间，
		 * 我们要求模板在新的默认
		 * 表空间中没有任何文件。这是必要的，因为否则被复制的
		 * 数据库将包含引用其默认
		 * 表空间的pg_class行，既包括显式引用（通过OID），
		 * 也包括隐式引用（作为零），这将导致问题。
		 * 例如，另一个使用复制数据库作为模板的CREATE DATABASE
		 * 操作，再次尝试更改其默认
		 * 表空间，将产生明确不正确的结果（它将
		 * 不当地将表移动到新的默认表空间，而这些表应
		 * 保持在同一表空间中）。
		 */
		if (fc_dst_deftablespace != fc_src_deftablespace)
		{
			char	   *fc_srcpath;
			struct stat fc_st;

			fc_srcpath = GetDatabasePath(fc_src_dboid, fc_dst_deftablespace);

			if (stat(fc_srcpath, &fc_st) == 0 &&
				S_ISDIR(fc_st.st_mode) &&
				!directory_is_empty(fc_srcpath))
				ereport(ERROR,
						(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
						 errmsg("cannot assign new default tablespace \"%s\"",
								fc_tablespacename),
						 errdetail("There is a conflict because database \"%s\" already has some tables in this tablespace.",
								   fc_dbtemplate)));
			pfree(fc_srcpath);
		}
	}
	else
	{
		/* 使用模板数据库的默认表空间 */
		fc_dst_deftablespace = fc_src_deftablespace;
		/* 注意在此路径中没有额外的权限检查 */
	}

	/*
	 * 如果使用了适当的开关，当违反数据库名称的回归测试
	 * 约定时发出警告。但在initdb期间不抱怨。
	 */
#ifdef ENFORCE_REGRESSION_TEST_NAME_RESTRICTIONS
	if (IsUnderPostmaster && strstr(fc_dbname, "regression") == NULL)
		elog(WARNING, "databases created by regression test cases should have names including \"regression\"");
#endif

	/*
	 * 检查数据库名称冲突。这只是为了提供比“唯一索引冲突”
	 * 更友好的错误消息。虽然存在竞争条件，但
	 * 在这种情况下我们愿意接受不那么友好的消息。
	 */
	if (OidIsValid(get_database_oid(fc_dbname, true)))
		ereport(ERROR,
				(errcode(ERRCODE_DUPLICATE_DATABASE),
				 errmsg("database \"%s\" already exists", fc_dbname)));

	/*
	 * 源数据库不能有任何活动的后端，除了这个
	 * （例外是允许在连接到template1时创建数据库）。
	 * 否则我们可能会复制不一致的数据。
	 *
	 * 这应该是基本错误检查中的最后一步，因为它涉及
	 * 潜在的等待；如果我们打算抛出错误，
	 * 不妨先抛出一个。
	 */
	if (CountOtherDBBackends(fc_src_dboid, &fc_notherbackends, &fc_npreparedxacts))
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_IN_USE),
				 errmsg("source database \"%s\" is being accessed by other users",
						fc_dbtemplate),
				 fc_errdetail_busy_db(fc_notherbackends, fc_npreparedxacts)));

	/*
	 * 为新数据库选择一个OID，检查它是否与
	 * 表空间目录中已有的任何文件名冲突。
	 */
	fc_pg_database_rel = table_open(DatabaseRelationId, RowExclusiveLock);

	/*
	 * 如果数据库OID已配置，检查该OID是否已在使用中或
	 * 数据目录是否已存在。
	 */
	if (OidIsValid(fc_dboid))
	{
		char	   *fc_existing_dbname = get_database_name(fc_dboid);

		if (fc_existing_dbname != NULL)
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE)),
					errmsg("database OID %u is already in use by database \"%s\"",
						   fc_dboid, fc_existing_dbname));

		if (fc_check_db_file_conflict(fc_dboid))
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE)),
					errmsg("data directory with the specified OID %u already exists", fc_dboid));
	}
	else
	{
		/* 如果未明确配置，则为新数据库选择一个OID。 */
		do
		{
			fc_dboid = GetNewOidWithIndex(fc_pg_database_rel, DatabaseOidIndexId,
									   Anum_pg_database_oid);
		} while (fc_check_db_file_conflict(fc_dboid));
	}

	/*
	 * 将一个新元组插入pg_database。这确立了我们对
	 * 新数据库名称的拥有权（任何其他尝试插入相同名称的人
	 * 都将在唯一索引上阻塞，并在我们提交后失败）。
	 */

	Assert((fc_dblocprovider == COLLPROVIDER_ICU && fc_dbiculocale) ||
		   (fc_dblocprovider != COLLPROVIDER_ICU && !fc_dbiculocale));

	/* 形成元组 */
	MemSet(fc_new_record, 0, sizeof(fc_new_record));
	MemSet(fc_new_record_nulls, false, sizeof(fc_new_record_nulls));

	fc_new_record[Anum_pg_database_oid - 1] = ObjectIdGetDatum(fc_dboid);
	fc_new_record[Anum_pg_database_datname - 1] =
		DirectFunctionCall1(namein, CStringGetDatum(fc_dbname));
	fc_new_record[Anum_pg_database_datdba - 1] = ObjectIdGetDatum(fc_datdba);
	fc_new_record[Anum_pg_database_encoding - 1] = Int32GetDatum(fc_encoding);
	fc_new_record[Anum_pg_database_datlocprovider - 1] = CharGetDatum(fc_dblocprovider);
	fc_new_record[Anum_pg_database_datistemplate - 1] = BoolGetDatum(fc_dbistemplate);
	fc_new_record[Anum_pg_database_datallowconn - 1] = BoolGetDatum(fc_dballowconnections);
	fc_new_record[Anum_pg_database_datconnlimit - 1] = Int32GetDatum(fc_dbconnlimit);
	fc_new_record[Anum_pg_database_datfrozenxid - 1] = TransactionIdGetDatum(fc_src_frozenxid);
	fc_new_record[Anum_pg_database_datminmxid - 1] = TransactionIdGetDatum(fc_src_minmxid);
	fc_new_record[Anum_pg_database_dattablespace - 1] = ObjectIdGetDatum(fc_dst_deftablespace);
	fc_new_record[Anum_pg_database_datcollate - 1] = CStringGetTextDatum(fc_dbcollate);
	fc_new_record[Anum_pg_database_datctype - 1] = CStringGetTextDatum(fc_dbctype);
	if (fc_dbiculocale)
		fc_new_record[Anum_pg_database_daticulocale - 1] = CStringGetTextDatum(fc_dbiculocale);
	else
		fc_new_record_nulls[Anum_pg_database_daticulocale - 1] = true;
	if (fc_dbcollversion)
		fc_new_record[Anum_pg_database_datcollversion - 1] = CStringGetTextDatum(fc_dbcollversion);
	else
		fc_new_record_nulls[Anum_pg_database_datcollversion - 1] = true;

	/*
	 * 我们故意将datacl设置为默认值（NULL），而不是从
	 * 模板数据库复制它。当所有者与模板的所有者不同时，
	 * 复制它将是一个坏主意。
	 */
	fc_new_record_nulls[Anum_pg_database_datacl - 1] = true;

	fc_tuple = heap_form_tuple(RelationGetDescr(fc_pg_database_rel),
							fc_new_record, fc_new_record_nulls);

	CatalogTupleInsert(fc_pg_database_rel, fc_tuple);

	/*
	 * 现在生成与新数据库相关的其他目录条目
	 */

	/* 注册所有者依赖关系 */
	recordDependencyOnOwner(DatabaseRelationId, fc_dboid, fc_datdba);

	/* 为数据库内的对象创建pg_shdepend条目 */
	copyTemplateDependencies(fc_src_dboid, fc_dboid);

	/* 新数据库的创建后钩子 */
	InvokeObjectPostCreateHook(DatabaseRelationId, fc_dboid, 0);

	/*
	 * 如果我们要将要创建的数据库的数据读取到
	 * shared_buffers 中，就需要对其加锁。没有人应该知道这个
	 * 数据库的存在，但保持一个 AccessExclusiveLock 锁住该
	 * 数据库的事实是好的，这样可以在不担心后续再读取更多数据的情况下
	 * 删除它的所有缓存。
	 *
	 * 请注意，我们需要在进入下面的
	 * PG_ENSURE_ERROR_CLEANUP 块之前这样做，因为 createdb_failure_callback
	 * 预期此锁已经被持有。
	 */
	if (fc_dbstrategy == CREATEDB_WAL_LOG)
		LockSharedObject(DatabaseRelationId, fc_dboid, 0, AccessShareLock);

	/*
	 * 一旦我们开始复制子目录，如果失败，我们需要能够清理它们。
	 * 使用 ENSURE 块来确保这种情况发生。（这并不是一个 100%
	 * 的解决方案，因为在我们离开这个例程后事务提交可能会失败，
	 * 但它应该可以处理大多数场景。）
	 */
	fc_fparms.src_dboid = fc_src_dboid;
	fc_fparms.dest_dboid = fc_dboid;
	fc_fparms.strategy = fc_dbstrategy;

	PG_ENSURE_ERROR_CLEANUP(fc_createdb_failure_callback,
							PointerGetDatum(&fc_fparms));
	{
		/*
		 * 如果用户要求使用 WAL_LOG 策略创建数据库，则调用
		 * CreateDatabaseUsingWalLog，这将按块级复制数据库，
		 * 并将每个复制的块进行 WAL 日志记录。
		 * 否则，调用 CreateDatabaseUsingFileCopy，将数据库
		 * 文件逐个复制。
		 */
		if (fc_dbstrategy == CREATEDB_WAL_LOG)
			fc_CreateDatabaseUsingWalLog(fc_src_dboid, fc_dboid, fc_src_deftablespace,
									  fc_dst_deftablespace);
		else
			fc_CreateDatabaseUsingFileCopy(fc_src_dboid, fc_dboid, fc_src_deftablespace,
										fc_dst_deftablespace);

		/*
		 * 关闭 pg_database，但在提交之前保持锁定。
		 */
		table_close(fc_pg_database_rel, NoLock);

		/*
		 * 强制同步提交，从而最小化数据库文件创建与事务提交之间的
		 * 时间窗口。如果我们在提交之前崩溃，我们将有一个占用磁盘
		 * 空间的数据库，但它不在 pg_database 中，这不好。
		 */
		ForceSyncCommit();
	}
	PG_END_ENSURE_ERROR_CLEANUP(fc_createdb_failure_callback,
								PointerGetDatum(&fc_fparms));

	return fc_dboid;
}

/*
 * 检查所选编码是否与所选区域设置匹配。此限制是必要的，
 * 因为 libc 的区域特定代码通常在遇到不期望的编码数据时会失败。
 * 我们在四种情况下允许不匹配：
 *
 * 1. 区域编码 = SQL_ASCII，这意味着区域设置为 C/POSIX，
 * 可与任何编码一起使用。
 *
 * 2. 区域编码 = -1，这意味着我们无法确定区域的编码，
 * 必须信任用户正确设置它。
 *
 * 3. 选定编码为 UTF8，且平台为 win32。这是因为
 * UTF8 是一个伪代码页，在所有区域中都得到支持，因为它
 * 在使用之前会被转换为 UTF16。
 *
 * 4. 选定编码为 SQL_ASCII，但仅在你是超级用户的情况下。
 * 这很冒险，但我们历史上允许它 --- 特别是，
 * 回归测试需要它。
 *
 * 注意：如果你更改了此策略，请修复 initdb 以匹配。
 */
void check_encoding_locale_matches(int fc_encoding, const char *fc_collate, const char *fc_ctype)
{
	int			fc_ctype_encoding = pg_get_encoding_from_locale(fc_ctype, true);
	int			fc_collate_encoding = pg_get_encoding_from_locale(fc_collate, true);

	if (!(fc_ctype_encoding == fc_encoding ||
		  fc_ctype_encoding == PG_SQL_ASCII ||
		  fc_ctype_encoding == -1 ||
#ifdef WIN32
		  encoding == PG_UTF8 ||
#endif
		  (fc_encoding == PG_SQL_ASCII && superuser())))
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("encoding \"%s\" does not match locale \"%s\"",
						pg_encoding_to_char(fc_encoding),
						fc_ctype),
				 errdetail("The chosen LC_CTYPE setting requires encoding \"%s\".",
						   pg_encoding_to_char(fc_ctype_encoding))));

	if (!(fc_collate_encoding == fc_encoding ||
		  fc_collate_encoding == PG_SQL_ASCII ||
		  fc_collate_encoding == -1 ||
#ifdef WIN32
		  encoding == PG_UTF8 ||
#endif
		  (fc_encoding == PG_SQL_ASCII && superuser())))
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("encoding \"%s\" does not match locale \"%s\"",
						pg_encoding_to_char(fc_encoding),
						fc_collate),
				 errdetail("The chosen LC_COLLATE setting requires encoding \"%s\".",
						   pg_encoding_to_char(fc_collate_encoding))));
}

/* 创建数据库的错误清理回调 */
static void fc_createdb_failure_callback(int fc_code, Datum fc_arg)
{
	createdb_failure_params *fc_fparms = (createdb_failure_params *) DatumGetPointer(fc_arg);

	/*
	 * 如果我们是按块级复制数据库，则删除共享缓冲区中目标
	 * 数据库的页面。并告诉检查点程序忘记数据库中
	 * 文件的任何待处理 fsync 和 unlink 请求。
	 * 做这个的原因与上面的 dropdb 函数解释的相同。
	 * 但与 dropdb 不同的是，我们不需要调用
	 * pgstat_drop_database，因为这个数据库仍然没有创建，
	 * 所以不应该有任何统计信息。
	 */
	if (fc_fparms->strategy == CREATEDB_WAL_LOG)
	{
		DropDatabaseBuffers(fc_fparms->dest_dboid);
		ForgetDatabaseSyncRequests(fc_fparms->dest_dboid);

		/* 释放目标数据库的锁。 */
		UnlockSharedObject(DatabaseRelationId, fc_fparms->dest_dboid, 0,
						   AccessShareLock);
	}

	/*
	 * 在执行递归删除之前释放源数据库的锁。这不是必需的，但尽快释放锁似乎是可取的。
	 */
	UnlockSharedObject(DatabaseRelationId, fc_fparms->src_dboid, 0, ShareLock);

	/* 丢弃任何成功复制的子目录 */
	fc_remove_dbtablespaces(fc_fparms->dest_dboid);
}


/*
 * 删除数据库
 */
void dropdb(const char *fc_dbname, bool fc_missing_ok, bool fc_force)
{
	Oid			fc_db_id;
	bool		fc_db_istemplate;
	Relation	fc_pgdbrel;
	HeapTuple	fc_tup;
	ScanKeyData fc_scankey;
	void	   *fc_inplace_state;
	Form_pg_database fc_datform;
	int			fc_notherbackends;
	int			fc_npreparedxacts;
	int			fc_nslots,
				fc_nslots_active;
	int			fc_nsubscriptions;

	/*
	 * 查找目标数据库的OID，并获得对其的独占锁。我们需要这样做，以确保在删除目标数据库时，没有新的后端在该数据库中启动（参见 postinit.c），并且没有人将其用作 CREATE DATABASE 模板或试图自行删除它。
	 */
	fc_pgdbrel = table_open(DatabaseRelationId, RowExclusiveLock);

	if (!fc_get_db_info(fc_dbname, AccessExclusiveLock, &fc_db_id, NULL, NULL,
					 &fc_db_istemplate, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL))
	{
		if (!fc_missing_ok)
		{
			ereport(ERROR,
					(errcode(ERRCODE_UNDEFINED_DATABASE),
					 errmsg("database \"%s\" does not exist", fc_dbname)));
		}
		else
		{
			/* 关闭 pg_database，释放锁，因为我们没有更改任何内容 */
			table_close(fc_pgdbrel, RowExclusiveLock);
			ereport(NOTICE,
					(errmsg("database \"%s\" does not exist, skipping",
							fc_dbname)));
			return;
		}
	}

	/*
	 * 权限检查
	 */
	if (!pg_database_ownercheck(fc_db_id, GetUserId()))
		aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
					   fc_dbname);

	/* 删除正在移除的数据库的钩子 */
	InvokeObjectDropHook(DatabaseRelationId, fc_db_id, 0);

	/*
	 * 禁止删除标记为 istemplate 的数据库。这只是为了防止人们不小心删除 template0 或 template1；如果他们真的下定决心，可以这样做……
	 */
	if (fc_db_istemplate)
		ereport(ERROR,
				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
				 errmsg("cannot drop a template database")));

	/* 显然不能删除我自己的数据库 */
	if (fc_db_id == MyDatabaseId)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_IN_USE),
				 errmsg("cannot drop the currently open database")));

	/*
	 * 检查是否有活动的逻辑槽引用要删除的数据库。我们持有的数据库锁防止使用数据库创建新槽或现有槽变为活动状态。
	 */
	(void) ReplicationSlotsCountDBSlots(fc_db_id, &fc_nslots, &fc_nslots_active);
	if (fc_nslots_active)
	{
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_IN_USE),
				 errmsg("database \"%s\" is used by an active logical replication slot",
						fc_dbname),
				 errdetail_plural("There is %d active slot.",
								  "There are %d active slots.",
								  fc_nslots_active, fc_nslots_active)));
	}

	/*
	 * 检查目标数据库中是否定义了订阅。
	 *
	 * 我们无法自动删除它们，因为它们可能在其他数据库/实例中占用资源。
	 */
	if ((fc_nsubscriptions = CountDBSubscriptions(fc_db_id)) > 0)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_IN_USE),
				 errmsg("database \"%s\" is being used by logical replication subscription",
						fc_dbname),
				 errdetail_plural("There is %d subscription.",
								  "There are %d subscriptions.",
								  fc_nsubscriptions, fc_nsubscriptions)));


	/*
	 * 如果用户请求，尝试终止所有现有连接到目标数据库的连接。
	 */
	if (fc_force)
		TerminateOtherDBBackends(fc_db_id);

	/*
	 * 检查目标数据库中的其他后端。（由于我们持有数据库锁，此后无法再启动新的后端。）
	 *
	 * 与 CREATE DATABASE 一样，在其他错误条件后检查这一点。
	 */
	if (CountOtherDBBackends(fc_db_id, &fc_notherbackends, &fc_npreparedxacts))
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_IN_USE),
				 errmsg("database \"%s\" is being accessed by other users",
						fc_dbname),
				 fc_errdetail_busy_db(fc_notherbackends, fc_npreparedxacts)));

	/*
	 * 删除与数据库相关的任何注释或安全标签。
	 */
	DeleteSharedComments(fc_db_id, DatabaseRelationId);
	DeleteSharedSecurityLabel(fc_db_id, DatabaseRelationId);

	/*
	 * 移除与此数据库相关的设置
	 */
	DropSetting(fc_db_id, InvalidOid);

	/*
	 * 移除数据库的共享依赖引用。
	 */
	dropDatabaseDependencies(fc_db_id);

	/*
	 * 告诉累积统计系统立即忘记它。
	 */
	pgstat_drop_database(fc_db_id);

	/*
	 * 除了删除目录行之外，后续操作不是事务性的（考虑 DropDatabaseBuffers() 丢弃已修改的缓冲区）。但我们可能会崩溃或在下面中断。为了防止访问内容无效的数据库，使用就地更新将数据库标记为无效。
	 *
	 * 我们需要在继续之前刷新 WAL，以确保修改是持久的，然后再执行不可逆的文件系统操作。
	 */
	ScanKeyInit(&fc_scankey,
				Anum_pg_database_datname,
				BTEqualStrategyNumber, F_NAMEEQ,
				CStringGetDatum(fc_dbname));
	systable_inplace_update_begin(fc_pgdbrel, DatabaseNameIndexId, true,
								  NULL, 1, &fc_scankey, &fc_tup, &fc_inplace_state);
	if (!HeapTupleIsValid(fc_tup))
		elog(ERROR, "cache lookup failed for database %u", fc_db_id);
	fc_datform = (Form_pg_database) GETSTRUCT(fc_tup);
	fc_datform->datconnlimit = DATCONNLIMIT_INVALID_DB;
	systable_inplace_update_finish(fc_inplace_state, fc_tup);
	XLogFlush(XactLastRecEnd);

	/*
	 * 也以事务性方式删除元组。如果该事务提交，该行将消失，但如果我们失败，dropdb() 可以再次调用。
	 */
	CatalogTupleDelete(fc_pgdbrel, &fc_tup->t_self);
	heap_freetuple(fc_tup);

	/*
	 * 删除数据库特定的复制槽。
	 */
	ReplicationSlotsDropDBSlots(fc_db_id);

	/*
	 * 删除此数据库在共享缓冲区中的页面。这
	 * 很重要，以确保没有剩余的后端试图稍后将脏缓冲区写出
	 * 到已经失效的数据库...
	 */
	DropDatabaseBuffers(fc_db_id);

	/*
	 * 告诉检查点器忘记数据库中
	 * 文件的任何挂起 fsync 和 unlink 请求；否则，fsync
	 * 将在下一个检查点时失败，或者更糟糕的是，它将删除属于
	 * 同一 OID 的新创建数据库的文件。
	 */
	ForgetDatabaseSyncRequests(fc_db_id);

	/*
	 * 强制执行检查点以确保检查点器已收到
	 * ForgetDatabaseSyncRequests 发送的消息。
	 */
	RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT);

	/* 关闭所有后端中的 smgr fds。 */
	WaitForProcSignalBarrier(EmitProcSignalBarrier(PROCSIGNAL_BARRIER_SMGRRELEASE));

	/*
	 * 移除属于该数据库的所有表空间子目录。
	 */
	fc_remove_dbtablespaces(fc_db_id);

	/*
	 * 关闭 pg_database，但在提交之前保持锁定。
	 */
	table_close(fc_pgdbrel, NoLock);

	/*
	 * 强制同步提交，从而最小化删除
	 * 数据库文件和事务提交之间的窗口。如果我们在
	 * 提交之前崩溃，我们将获得一个在磁盘上消失但在
	 * pg_database 中仍然存在的数据库，这是不好的。
	 */
	ForceSyncCommit();
}


/*
 * 重命名数据库
 */
ObjectAddress RenameDatabase(const char *fc_oldname, const char *fc_newname)
{
	Oid			fc_db_id;
	HeapTuple	fc_newtup;
	ItemPointerData fc_otid;
	Relation	fc_rel;
	int			fc_notherbackends;
	int			fc_npreparedxacts;
	ObjectAddress fc_address;

	/*
	 * 查找目标数据库的 OID，并在其上获取排他性锁定。我们
	 * 需要这样做的原因与 DROP DATABASE 相同。
	 */
	fc_rel = table_open(DatabaseRelationId, RowExclusiveLock);

	if (!fc_get_db_info(fc_oldname, AccessExclusiveLock, &fc_db_id, NULL, NULL,
					 NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_DATABASE),
				 errmsg("database \"%s\" does not exist", fc_oldname)));

	/* 必须是所有者 */
	if (!pg_database_ownercheck(fc_db_id, GetUserId()))
		aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
					   fc_oldname);

	/* 必须具有创建数据库权限 */
	if (!fc_have_createdb_privilege())
		ereport(ERROR,
				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
				 errmsg("permission denied to rename database")));

	/*
	 * 如果在构建时带有适当的开关，当回归测试
	 * 数据库名称约定被违反时会发出警告。
	 */
#ifdef ENFORCE_REGRESSION_TEST_NAME_RESTRICTIONS
	if (strstr(fc_newname, "regression") == NULL)
		elog(WARNING, "databases created by regression test cases should have names including \"regression\"");
#endif

	/*
	 * 确保新名称不存在。有关 CREATE DATABASE 中相同错误的说明。
	 */
	if (OidIsValid(get_database_oid(fc_newname, true)))
		ereport(ERROR,
				(errcode(ERRCODE_DUPLICATE_DATABASE),
				 errmsg("database \"%s\" already exists", fc_newname)));

	/*
	 * XXX 客户端应用程序可能会在某处存储当前数据库，
	 * 因此重命名它可能会造成混淆。另一方面，除了
	 * 一点混淆外，可能没有实际问题，因此考虑一下
	 * 并做出决定。
	 */
	if (fc_db_id == MyDatabaseId)
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("current database cannot be renamed")));

	/*
	 * 确保数据库没有活动会话。这是与上述情况相同的
	 * 担忧，但适用于其他会话。
	 *
	 * 与 CREATE DATABASE 一样，在其他错误条件之后检查此项。
	 */
	if (CountOtherDBBackends(fc_db_id, &fc_notherbackends, &fc_npreparedxacts))
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_IN_USE),
				 errmsg("database \"%s\" is being accessed by other users",
						fc_oldname),
				 fc_errdetail_busy_db(fc_notherbackends, fc_npreparedxacts)));

	/* 重命名 */
	fc_newtup = SearchSysCacheLockedCopy1(DATABASEOID, ObjectIdGetDatum(fc_db_id));
	if (!HeapTupleIsValid(fc_newtup))
		elog(ERROR, "cache lookup failed for database %u", fc_db_id);
	fc_otid = fc_newtup->t_self;
	namestrcpy(&(((Form_pg_database) GETSTRUCT(fc_newtup))->datname), fc_newname);
	CatalogTupleUpdate(fc_rel, &fc_otid, fc_newtup);
	UnlockTuple(fc_rel, &fc_otid, InplaceUpdateTupleLock);

	InvokeObjectPostAlterHook(DatabaseRelationId, fc_db_id, 0);

	ObjectAddressSet(fc_address, DatabaseRelationId, fc_db_id);

	/*
	 * 关闭 pg_database，但在提交之前保持锁定。
	 */
	table_close(fc_rel, NoLock);

	return fc_address;
}


/*
 * ALTER DATABASE SET TABLESPACE
 */
static void fc_movedb(const char *fc_dbname, const char *fc_tblspcname)
{
	Oid			fc_db_id;
	Relation	fc_pgdbrel;
	int			fc_notherbackends;
	int			fc_npreparedxacts;
	HeapTuple	fc_oldtuple,
				fc_newtuple;
	Oid			fc_src_tblspcoid,
				fc_dst_tblspcoid;
	Datum		fc_new_record[Natts_pg_database];
	bool		fc_new_record_nulls[Natts_pg_database];
	bool		fc_new_record_repl[Natts_pg_database];
	ScanKeyData fc_scankey;
	SysScanDesc fc_sysscan;
	AclResult	fc_aclresult;
	char	   *fc_src_dbpath;
	char	   *fc_dst_dbpath;
	DIR		   *fc_dstdir;
	struct dirent *fc_xlde;
	movedb_failure_params fc_fparms;

	/*
	 * 查找目标数据库的 OID，并在其上获取排他性锁定。我们
	 * 需要这样做，以确保在移动数据库时没有新的后端启动，且
	 * 没有人将其用作 CREATE DATABASE 模板或尝试删除它。
	 */
	fc_pgdbrel = table_open(DatabaseRelationId, RowExclusiveLock);

	if (!fc_get_db_info(fc_dbname, AccessExclusiveLock, &fc_db_id, NULL, NULL,
					 NULL, NULL, NULL, NULL, &fc_src_tblspcoid, NULL, NULL, NULL, NULL, NULL))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_DATABASE),
				 errmsg("database \"%s\" does not exist", fc_dbname)));

	/*
	 * 我们实际上需要一个会话锁，以便锁定在
	 * 以下提交/重启中保持存在。（在提交时锁定几乎可以释放，
	 * 除非有人试图将数据库的关系移动回旧目录，而我们正在
	 * rmtree() 它。）
	 */
	LockSharedObjectForSession(DatabaseRelationId, fc_db_id, 0,
							   AccessExclusiveLock);

	/*
	 * 权限检查
	 */
	if (!pg_database_ownercheck(fc_db_id, GetUserId()))
		aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
					   fc_dbname);

	/*
	 * 显然不能移动我自己的数据库的表。
	 */
	if (fc_db_id == MyDatabaseId)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_IN_USE),
				 errmsg("cannot change the tablespace of the currently open database")));

	/*
	 * 获取表空间的 oid
	 */
	fc_dst_tblspcoid = get_tablespace_oid(fc_tblspcname, false);

	/*
	 * 权限检查
	 */
	fc_aclresult = pg_tablespace_aclcheck(fc_dst_tblspcoid, GetUserId(),
									   ACL_CREATE);
	if (fc_aclresult != ACLCHECK_OK)
		aclcheck_error(fc_aclresult, OBJECT_TABLESPACE,
					   fc_tblspcname);

	/*
	 * pg_global 绝不应是默认表空间
	 */
	if (fc_dst_tblspcoid == GLOBALTABLESPACE_OID)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("pg_global cannot be used as default tablespace")));

	/*
	 * 如果是同一表空间则无操作
	 */
	if (fc_src_tblspcoid == fc_dst_tblspcoid)
	{
		table_close(fc_pgdbrel, NoLock);
		UnlockSharedObjectForSession(DatabaseRelationId, fc_db_id, 0,
									 AccessExclusiveLock);
		return;
	}

	/*
	 * 检查目标数据库中的其他后端。（由于我们持有数据库锁，此后无法再启动新的后端。）
	 *
	 * 与 CREATE DATABASE 一样，在其他错误条件后检查这一点。
	 */
	if (CountOtherDBBackends(fc_db_id, &fc_notherbackends, &fc_npreparedxacts))
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_IN_USE),
				 errmsg("database \"%s\" is being accessed by other users",
						fc_dbname),
				 fc_errdetail_busy_db(fc_notherbackends, fc_npreparedxacts)));

	/*
	 * 获取旧的和新的数据库路径
	 */
	fc_src_dbpath = GetDatabasePath(fc_db_id, fc_src_tblspcoid);
	fc_dst_dbpath = GetDatabasePath(fc_db_id, fc_dst_tblspcoid);

	/*
	 * 在继续之前强制进行检查点。这将强制所有脏缓冲区，包括未记录表的脏缓冲区，刷新到磁盘，以确保源数据库在复制时在磁盘上是最新的。
	 * FlushDatabaseBuffers() 就足够了，但我们还希望处理任何待处理的删除请求。否则，在目标目录中检查现有文件可能会不必要地失败，更不用说由于源文件在其下被删除而导致复制失败。
	 * 在 Windows 上，这还确保后台进程不会持有任何打开的文件，这会导致 rmdir() 失败。
	 */
	RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT
					  | CHECKPOINT_FLUSH_ALL);

	/* 关闭所有后端中的 smgr fds。 */
	WaitForProcSignalBarrier(EmitProcSignalBarrier(PROCSIGNAL_BARRIER_SMGRRELEASE));

	/*
	 * 现在丢弃所有持有目标数据库数据的缓冲区；它们不应再是脏的，因此 DropDatabaseBuffers 是安全的。
	 *
	 * 似乎我们可以让这些缓冲区自然地从共享缓冲区中老化，因为它们不应再被引用。 问题在于，如果用户稍后将数据库移动回其原始表空间，任何仍然存活的缓冲区将显示包含有效数据——但它们将缺少在新表空间中进行的任何更改。 总之，释放那些不应再使用的缓冲区似乎是值得的。
	 *
	 * 注意：只需清除与 db_id 和 src_tblspcoid 匹配的缓冲区就足够了，但 bufmgr.c 目前没有为此提供 API。
	 */
	DropDatabaseBuffers(fc_db_id);

	/*
	 * 检查目标目录中是否存在文件，即此数据库的对象是否已经在目标表空间中。 我们不能允许在这种情况下移动，因为我们需要将这些关系的 pg_class.reltablespace 条目更改为零，而我们无法访问数据库的 pg_class 来完成此操作。
	 */
	fc_dstdir = AllocateDir(fc_dst_dbpath);
	if (fc_dstdir != NULL)
	{
		while ((fc_xlde = ReadDir(fc_dstdir, fc_dst_dbpath)) != NULL)
		{
			if (strcmp(fc_xlde->d_name, ".") == 0 ||
				strcmp(fc_xlde->d_name, "..") == 0)
				continue;

			ereport(ERROR,
					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
					 errmsg("some relations of database \"%s\" are already in tablespace \"%s\"",
							fc_dbname, fc_tblspcname),
					 errhint("You must move them back to the database's default tablespace before using this command.")));
		}

		FreeDir(fc_dstdir);

		/*
		 * 目录存在但为空。我们必须在使用 copydir 函数之前删除它。
		 */
		if (rmdir(fc_dst_dbpath) != 0)
			elog(ERROR, "could not remove directory \"%s\": %m",
				 fc_dst_dbpath);
	}

	/*
	 * 使用 ENSURE 块确保在复制失败时（例如，由于磁盘空间不足）删除残余物。这不是 100% 的解决方案，因为在事务提交期间可能会失败，但它应该能处理大多数情况。
	 */
	fc_fparms.dest_dboid = fc_db_id;
	fc_fparms.dest_tsoid = fc_dst_tblspcoid;
	PG_ENSURE_ERROR_CLEANUP(fc_movedb_failure_callback,
							PointerGetDatum(&fc_fparms));
	{
		/*
		 * 将文件从旧表空间复制到新表空间
		 */
		copydir(fc_src_dbpath, fc_dst_dbpath, false);

		/*
		 * 在 XLOG 中记录文件系统更改
		 */
		{
			xl_dbase_create_file_copy_rec fc_xlrec;

			fc_xlrec.db_id = fc_db_id;
			fc_xlrec.tablespace_id = fc_dst_tblspcoid;
			fc_xlrec.src_db_id = fc_db_id;
			fc_xlrec.src_tablespace_id = fc_src_tblspcoid;

			XLogBeginInsert();
			XLogRegisterData((char *) &fc_xlrec,
							 sizeof(xl_dbase_create_file_copy_rec));

			(void) XLogInsert(RM_DBASE_ID,
							  XLOG_DBASE_CREATE_FILE_COPY | XLR_SPECIAL_REL_UPDATE);
		}

		/*
		 * 更新数据库的 pg_database 元组
		 */
		ScanKeyInit(&fc_scankey,
					Anum_pg_database_datname,
					BTEqualStrategyNumber, F_NAMEEQ,
					CStringGetDatum(fc_dbname));
		fc_sysscan = systable_beginscan(fc_pgdbrel, DatabaseNameIndexId, true,
									 NULL, 1, &fc_scankey);
		fc_oldtuple = systable_getnext(fc_sysscan);
		if (!HeapTupleIsValid(fc_oldtuple))	/* 不应该发生... */
			ereport(ERROR,
					(errcode(ERRCODE_UNDEFINED_DATABASE),
					 errmsg("database \"%s\" does not exist", fc_dbname)));
		LockTuple(fc_pgdbrel, &fc_oldtuple->t_self, InplaceUpdateTupleLock);

		MemSet(fc_new_record, 0, sizeof(fc_new_record));
		MemSet(fc_new_record_nulls, false, sizeof(fc_new_record_nulls));
		MemSet(fc_new_record_repl, false, sizeof(fc_new_record_repl));

		fc_new_record[Anum_pg_database_dattablespace - 1] = ObjectIdGetDatum(fc_dst_tblspcoid);
		fc_new_record_repl[Anum_pg_database_dattablespace - 1] = true;

		fc_newtuple = heap_modify_tuple(fc_oldtuple, RelationGetDescr(fc_pgdbrel),
									 fc_new_record,
									 fc_new_record_nulls, fc_new_record_repl);
		CatalogTupleUpdate(fc_pgdbrel, &fc_oldtuple->t_self, fc_newtuple);
		UnlockTuple(fc_pgdbrel, &fc_oldtuple->t_self, InplaceUpdateTupleLock);

		InvokeObjectPostAlterHook(DatabaseRelationId, fc_db_id, 0);

		systable_endscan(fc_sysscan);

		/*
		 * 在此强制进行另一次检查点。 和 CREATE DATABASE 一样，这是为了确保我们不必重新执行已提交的 XLOG_DBASE_CREATE_FILE_COPY 操作，这将导致我们在下一个检查点之前丢失在新数据库表空间中进行的任何未记录操作。
		 */
		RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT);

		/*
		 * 强制同步提交，从而最小化复制数据库文件和事务提交之间的窗口。如果我们在提交之前崩溃，将会在磁盘上留下一个孤立的文件集，这并不是致命的，但也不好。
		 */
		ForceSyncCommit();

		/*
		 * 关闭 pg_database，但在提交之前保持锁定。
		 */
		table_close(fc_pgdbrel, NoLock);
	}
	PG_END_ENSURE_ERROR_CLEANUP(fc_movedb_failure_callback,
								PointerGetDatum(&fc_fparms));

	/*
	 * 提交事务，以便 pg_database 更新被提交。如果我们在删除文件时崩溃，数据库不会损坏，只会在旧目录中留下些孤立的文件。
	 *
	 * （这样是可以的，因为我们知道我们不在事务块内。）
	 *
	 * XXX 在确保块内这样做安全/更好么？不确定这是个好主意；考虑在事务真正提交后立即 elog。
	 */
	PopActiveSnapshot();
	CommitTransactionCommand();

	/* 为剩余工作开始新事务；不需要快照 */
	StartTransactionCommand();

	/*
	 * 从旧表空间中移除文件
	 */
	if (!rmtree(fc_src_dbpath, true))
		ereport(WARNING,
				(errmsg("some useless files may be left behind in old database directory \"%s\"",
						fc_src_dbpath)));

	/*
	 * 在 XLOG 中记录文件系统更改
	 */
	{
		xl_dbase_drop_rec fc_xlrec;

		fc_xlrec.db_id = fc_db_id;
		fc_xlrec.ntablespaces = 1;

		XLogBeginInsert();
		XLogRegisterData((char *) &fc_xlrec, sizeof(xl_dbase_drop_rec));
		XLogRegisterData((char *) &fc_src_tblspcoid, sizeof(Oid));

		(void) XLogInsert(RM_DBASE_ID,
						  XLOG_DBASE_DROP | XLR_SPECIAL_REL_UPDATE);
	}

	/* 现在可以安全地释放数据库锁 */
	UnlockSharedObjectForSession(DatabaseRelationId, fc_db_id, 0,
								 AccessExclusiveLock);

	pfree(fc_src_dbpath);
	pfree(fc_dst_dbpath);
}

/* movedb 的错误清理回调 */
static void fc_movedb_failure_callback(int fc_code, Datum fc_arg)
{
	movedb_failure_params *fc_fparms = (movedb_failure_params *) DatumGetPointer(fc_arg);
	char	   *fc_dstpath;

	/* 处理我们成功复制到目标目录的任何内容 */
	fc_dstpath = GetDatabasePath(fc_fparms->dest_dboid, fc_fparms->dest_tsoid);

	(void) rmtree(fc_dstpath, true);

	pfree(fc_dstpath);
}

/*
 * 处理选项并调用 dropdb 函数。
 */
void DropDatabase(ParseState *fc_pstate, DropdbStmt *fc_stmt)
{
	bool		fc_force = false;
	ListCell   *fc_lc;

	foreach(fc_lc, fc_stmt->options)
	{
		DefElem    *fc_opt = (DefElem *) lfirst(fc_lc);

		if (strcmp(fc_opt->defname, "force") == 0)
			fc_force = true;
		else
			ereport(ERROR,
					(errcode(ERRCODE_SYNTAX_ERROR),
					 errmsg("unrecognized DROP DATABASE option \"%s\"", fc_opt->defname),
					 parser_errposition(fc_pstate, fc_opt->location)));
	}

	dropdb(fc_stmt->dbname, fc_stmt->missing_ok, fc_force);
}

/*
 * ALTER DATABASE name ...
 */
Oid AlterDatabase(ParseState *fc_pstate, AlterDatabaseStmt *fc_stmt, bool fc_isTopLevel)
{
	Relation	fc_rel;
	Oid			fc_dboid;
	HeapTuple	fc_tuple,
				fc_newtuple;
	Form_pg_database fc_datform;
	ScanKeyData fc_scankey;
	SysScanDesc fc_scan;
	ListCell   *fc_option;
	bool		fc_dbistemplate = false;
	bool		fc_dballowconnections = true;
	int			fc_dbconnlimit = DATCONNLIMIT_UNLIMITED;
	DefElem    *fc_distemplate = NULL;
	DefElem    *fc_dallowconnections = NULL;
	DefElem    *fc_dconnlimit = NULL;
	DefElem    *fc_dtablespace = NULL;
	Datum		fc_new_record[Natts_pg_database];
	bool		fc_new_record_nulls[Natts_pg_database];
	bool		fc_new_record_repl[Natts_pg_database];

	/* 从语句节点树中提取选项 */
	foreach(fc_option, fc_stmt->options)
	{
		DefElem    *fc_defel = (DefElem *) lfirst(fc_option);

		if (strcmp(fc_defel->defname, "is_template") == 0)
		{
			if (fc_distemplate)
				errorConflictingDefElem(fc_defel, fc_pstate);
			fc_distemplate = fc_defel;
		}
		else if (strcmp(fc_defel->defname, "allow_connections") == 0)
		{
			if (fc_dallowconnections)
				errorConflictingDefElem(fc_defel, fc_pstate);
			fc_dallowconnections = fc_defel;
		}
		else if (strcmp(fc_defel->defname, "connection_limit") == 0)
		{
			if (fc_dconnlimit)
				errorConflictingDefElem(fc_defel, fc_pstate);
			fc_dconnlimit = fc_defel;
		}
		else if (strcmp(fc_defel->defname, "tablespace") == 0)
		{
			if (fc_dtablespace)
				errorConflictingDefElem(fc_defel, fc_pstate);
			fc_dtablespace = fc_defel;
		}
		else
			ereport(ERROR,
					(errcode(ERRCODE_SYNTAX_ERROR),
					 errmsg("option \"%s\" not recognized", fc_defel->defname),
					 parser_errposition(fc_pstate, fc_defel->location)));
	}

	if (fc_dtablespace)
	{
		/*
		 * 虽然 SET TABLESPACE 语法不允许任何其他选项，但有人可能会写 "WITH TABLESPACE ...”。在这种情况下禁止指定任何其他选项。
		 */
		if (list_length(fc_stmt->options) != 1)
			ereport(ERROR,
					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
					 errmsg("option \"%s\" cannot be specified with other options",
							fc_dtablespace->defname),
					 parser_errposition(fc_pstate, fc_dtablespace->location)));
		/* 这个情况在事务块内是不允许的 */
		PreventInTransactionBlock(fc_isTopLevel, "ALTER DATABASE SET TABLESPACE");
		fc_movedb(fc_stmt->dbname, defGetString(fc_dtablespace));
		return InvalidOid;
	}

	if (fc_distemplate && fc_distemplate->arg)
		fc_dbistemplate = defGetBoolean(fc_distemplate);
	if (fc_dallowconnections && fc_dallowconnections->arg)
		fc_dballowconnections = defGetBoolean(fc_dallowconnections);
	if (fc_dconnlimit && fc_dconnlimit->arg)
	{
		fc_dbconnlimit = defGetInt32(fc_dconnlimit);
		if (fc_dbconnlimit < DATCONNLIMIT_UNLIMITED)
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("invalid connection limit: %d", fc_dbconnlimit)));
	}

	/*
	 * 获取旧元组。我们不需要对数据库锁定，
	 * 因为我们不会做任何会干扰传入连接的事情。
	 */
	fc_rel = table_open(DatabaseRelationId, RowExclusiveLock);
	ScanKeyInit(&fc_scankey,
				Anum_pg_database_datname,
				BTEqualStrategyNumber, F_NAMEEQ,
				CStringGetDatum(fc_stmt->dbname));
	fc_scan = systable_beginscan(fc_rel, DatabaseNameIndexId, true,
							  NULL, 1, &fc_scankey);
	fc_tuple = systable_getnext(fc_scan);
	if (!HeapTupleIsValid(fc_tuple))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_DATABASE),
				 errmsg("database \"%s\" does not exist", fc_stmt->dbname)));
	LockTuple(fc_rel, &fc_tuple->t_self, InplaceUpdateTupleLock);

	fc_datform = (Form_pg_database) GETSTRUCT(fc_tuple);
	fc_dboid = fc_datform->oid;

	if (database_is_invalid_form(fc_datform))
	{
		ereport(FATAL,
				errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				errmsg("cannot alter invalid database \"%s\"", fc_stmt->dbname),
				errhint("Use DROP DATABASE to drop invalid databases."));
	}

	if (!pg_database_ownercheck(fc_dboid, GetUserId()))
		aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
					   fc_stmt->dbname);

	/*
	 * 为了避免被锁定并不得不进入独立模式，我们拒绝禁止对当前连接的数据库的连接。
	 * 锁定仍然可以发生，但并发会话的这种可能性并不足以让人担忧。
	 */
	if (!fc_dballowconnections && fc_dboid == MyDatabaseId)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("cannot disallow connections for current database")));

	/*
	 * 构建更新的元组，浏览刚获得的信息
	 */
	MemSet(fc_new_record, 0, sizeof(fc_new_record));
	MemSet(fc_new_record_nulls, false, sizeof(fc_new_record_nulls));
	MemSet(fc_new_record_repl, false, sizeof(fc_new_record_repl));

	if (fc_distemplate)
	{
		fc_new_record[Anum_pg_database_datistemplate - 1] = BoolGetDatum(fc_dbistemplate);
		fc_new_record_repl[Anum_pg_database_datistemplate - 1] = true;
	}
	if (fc_dallowconnections)
	{
		fc_new_record[Anum_pg_database_datallowconn - 1] = BoolGetDatum(fc_dballowconnections);
		fc_new_record_repl[Anum_pg_database_datallowconn - 1] = true;
	}
	if (fc_dconnlimit)
	{
		fc_new_record[Anum_pg_database_datconnlimit - 1] = Int32GetDatum(fc_dbconnlimit);
		fc_new_record_repl[Anum_pg_database_datconnlimit - 1] = true;
	}

	fc_newtuple = heap_modify_tuple(fc_tuple, RelationGetDescr(fc_rel), fc_new_record,
								 fc_new_record_nulls, fc_new_record_repl);
	CatalogTupleUpdate(fc_rel, &fc_tuple->t_self, fc_newtuple);
	UnlockTuple(fc_rel, &fc_tuple->t_self, InplaceUpdateTupleLock);

	InvokeObjectPostAlterHook(DatabaseRelationId, fc_dboid, 0);

	systable_endscan(fc_scan);

	/* 关闭 pg_database，但在提交之前保持锁定 */
	table_close(fc_rel, NoLock);

	return fc_dboid;
}


/*
 * ALTER DATABASE name REFRESH COLLATION VERSION
 */
ObjectAddress AlterDatabaseRefreshColl(AlterDatabaseRefreshCollStmt *fc_stmt)
{
	Relation	fc_rel;
	ScanKeyData fc_scankey;
	SysScanDesc fc_scan;
	Oid			fc_db_id;
	HeapTuple	fc_tuple;
	Form_pg_database fc_datForm;
	ObjectAddress fc_address;
	Datum		fc_datum;
	bool		fc_isnull;
	char	   *fc_oldversion;
	char	   *fc_newversion;

	fc_rel = table_open(DatabaseRelationId, RowExclusiveLock);
	ScanKeyInit(&fc_scankey,
				Anum_pg_database_datname,
				BTEqualStrategyNumber, F_NAMEEQ,
				CStringGetDatum(fc_stmt->dbname));
	fc_scan = systable_beginscan(fc_rel, DatabaseNameIndexId, true,
							  NULL, 1, &fc_scankey);
	fc_tuple = systable_getnext(fc_scan);
	if (!HeapTupleIsValid(fc_tuple))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_DATABASE),
				 errmsg("database \"%s\" does not exist", fc_stmt->dbname)));

	fc_datForm = (Form_pg_database) GETSTRUCT(fc_tuple);
	fc_db_id = fc_datForm->oid;

	if (!pg_database_ownercheck(fc_db_id, GetUserId()))
		aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
					   fc_stmt->dbname);
	LockTuple(fc_rel, &fc_tuple->t_self, InplaceUpdateTupleLock);

	fc_datum = heap_getattr(fc_tuple, Anum_pg_database_datcollversion, RelationGetDescr(fc_rel), &fc_isnull);
	fc_oldversion = fc_isnull ? NULL : TextDatumGetCString(fc_datum);

	fc_datum = heap_getattr(fc_tuple, fc_datForm->datlocprovider == COLLPROVIDER_ICU ? Anum_pg_database_daticulocale : Anum_pg_database_datcollate, RelationGetDescr(fc_rel), &fc_isnull);
	if (fc_isnull)
		elog(ERROR, "unexpected null in pg_database");
	fc_newversion = get_collation_actual_version(fc_datForm->datlocprovider, TextDatumGetCString(fc_datum));

	/* 不能从 NULL 改为非 NULL 或反之 */
	if ((!fc_oldversion && fc_newversion) || (fc_oldversion && !fc_newversion))
		elog(ERROR, "invalid collation version change");
	else if (fc_oldversion && fc_newversion && strcmp(fc_newversion, fc_oldversion) != 0)
	{
		bool		fc_nulls[Natts_pg_database] = {0};
		bool		fc_replaces[Natts_pg_database] = {0};
		Datum		fc_values[Natts_pg_database] = {0};
		HeapTuple	fc_newtuple;

		ereport(NOTICE,
				(errmsg("changing version from %s to %s",
						fc_oldversion, fc_newversion)));

		fc_values[Anum_pg_database_datcollversion - 1] = CStringGetTextDatum(fc_newversion);
		fc_replaces[Anum_pg_database_datcollversion - 1] = true;

		fc_newtuple = heap_modify_tuple(fc_tuple, RelationGetDescr(fc_rel),
									 fc_values, fc_nulls, fc_replaces);
		CatalogTupleUpdate(fc_rel, &fc_tuple->t_self, fc_newtuple);
		heap_freetuple(fc_newtuple);
	}
	else
		ereport(NOTICE,
				(errmsg("version has not changed")));
	UnlockTuple(fc_rel, &fc_tuple->t_self, InplaceUpdateTupleLock);

	InvokeObjectPostAlterHook(DatabaseRelationId, fc_db_id, 0);

	ObjectAddressSet(fc_address, DatabaseRelationId, fc_db_id);

	systable_endscan(fc_scan);

	table_close(fc_rel, NoLock);

	return fc_address;
}


/*
 * ALTER DATABASE name SET ...
 */
Oid AlterDatabaseSet(AlterDatabaseSetStmt *fc_stmt)
{
	Oid			fc_datid = get_database_oid(fc_stmt->dbname, false);

	/*
	 * 获取数据库的锁，并确保它在此期间没有消失。
	 */
	shdepLockAndCheckObject(DatabaseRelationId, fc_datid);

	if (!pg_database_ownercheck(fc_datid, GetUserId()))
		aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
					   fc_stmt->dbname);

	AlterSetting(fc_datid, InvalidOid, fc_stmt->setstmt);

	UnlockSharedObject(DatabaseRelationId, fc_datid, 0, AccessShareLock);

	return fc_datid;
}


/*
 * ALTER DATABASE name OWNER TO newowner
 */
ObjectAddress AlterDatabaseOwner(const char *fc_dbname, Oid fc_newOwnerId)
{
	Oid			fc_db_id;
	HeapTuple	fc_tuple;
	Relation	fc_rel;
	ScanKeyData fc_scankey;
	SysScanDesc fc_scan;
	Form_pg_database fc_datForm;
	ObjectAddress fc_address;

	/*
	 * 获取旧元组。我们不需要对数据库锁定，
	 * 因为我们不会做任何会干扰传入连接的事情。
	 */
	fc_rel = table_open(DatabaseRelationId, RowExclusiveLock);
	ScanKeyInit(&fc_scankey,
				Anum_pg_database_datname,
				BTEqualStrategyNumber, F_NAMEEQ,
				CStringGetDatum(fc_dbname));
	fc_scan = systable_beginscan(fc_rel, DatabaseNameIndexId, true,
							  NULL, 1, &fc_scankey);
	fc_tuple = systable_getnext(fc_scan);
	if (!HeapTupleIsValid(fc_tuple))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_DATABASE),
				 errmsg("database \"%s\" does not exist", fc_dbname)));

	fc_datForm = (Form_pg_database) GETSTRUCT(fc_tuple);
	fc_db_id = fc_datForm->oid;

	/*
	 * 如果新所有者与现有所有者相同，则认为命令已成功。这是为了与其他对象保持一致。
	 */
	if (fc_datForm->datdba != fc_newOwnerId)
	{
		Datum		fc_repl_val[Natts_pg_database];
		bool		fc_repl_null[Natts_pg_database];
		bool		fc_repl_repl[Natts_pg_database];
		Acl		   *fc_newAcl;
		Datum		fc_aclDatum;
		bool		fc_isNull;
		HeapTuple	fc_newtuple;

		/* 否则，必须是现有对象的所有者 */
		if (!pg_database_ownercheck(fc_db_id, GetUserId()))
			aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_DATABASE,
						   fc_dbname);

		/* 必须能够成为新所有者 */
		check_is_member_of_role(GetUserId(), fc_newOwnerId);

		/*
		 * 必须具有创建数据库权限
		 *
		 * 注意：这与其他更改所有者检查不同，因为当前用户被检查是否具有创建数据库的权限，而不是目标所有者。这与数据库的创建案例是一致的。由于超级用户将始终拥有此权限，我们不需要为他们特殊处理。
		 */
		if (!fc_have_createdb_privilege())
			ereport(ERROR,
					(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
					 errmsg("permission denied to change owner of database")));

		LockTuple(fc_rel, &fc_tuple->t_self, InplaceUpdateTupleLock);

		memset(fc_repl_null, false, sizeof(fc_repl_null));
		memset(fc_repl_repl, false, sizeof(fc_repl_repl));

		fc_repl_repl[Anum_pg_database_datdba - 1] = true;
		fc_repl_val[Anum_pg_database_datdba - 1] = ObjectIdGetDatum(fc_newOwnerId);

		/*
		 * 确定新所有者的修改后 ACL。这仅在 ACL 非空时必要。
		 */
		fc_aclDatum = heap_getattr(fc_tuple,
								Anum_pg_database_datacl,
								RelationGetDescr(fc_rel),
								&fc_isNull);
		if (!fc_isNull)
		{
			fc_newAcl = aclnewowner(DatumGetAclP(fc_aclDatum),
								 fc_datForm->datdba, fc_newOwnerId);
			fc_repl_repl[Anum_pg_database_datacl - 1] = true;
			fc_repl_val[Anum_pg_database_datacl - 1] = PointerGetDatum(fc_newAcl);
		}

		fc_newtuple = heap_modify_tuple(fc_tuple, RelationGetDescr(fc_rel), fc_repl_val, fc_repl_null, fc_repl_repl);
		CatalogTupleUpdate(fc_rel, &fc_newtuple->t_self, fc_newtuple);
		UnlockTuple(fc_rel, &fc_tuple->t_self, InplaceUpdateTupleLock);

		heap_freetuple(fc_newtuple);

		/* 更新所有者依赖关系引用 */
		changeDependencyOnOwner(DatabaseRelationId, fc_db_id, fc_newOwnerId);
	}

	InvokeObjectPostAlterHook(DatabaseRelationId, fc_db_id, 0);

	ObjectAddressSet(fc_address, DatabaseRelationId, fc_db_id);

	systable_endscan(fc_scan);

	/* 关闭 pg_database，但在提交之前保持锁定 */
	table_close(fc_rel, NoLock);

	return fc_address;
}


Datum pg_database_collation_actual_version(PG_FUNCTION_ARGS)
{
	Oid			fc_dbid = PG_GETARG_OID(0);
	HeapTuple	fc_tp;
	char		fc_datlocprovider;
	Datum		fc_datum;
	bool		fc_isnull;
	char	   *fc_version;

	fc_tp = SearchSysCache1(DATABASEOID, ObjectIdGetDatum(fc_dbid));
	if (!HeapTupleIsValid(fc_tp))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("database with OID %u does not exist", fc_dbid)));

	fc_datlocprovider = ((Form_pg_database) GETSTRUCT(fc_tp))->datlocprovider;

	fc_datum = SysCacheGetAttr(DATABASEOID, fc_tp, fc_datlocprovider == COLLPROVIDER_ICU ? Anum_pg_database_daticulocale : Anum_pg_database_datcollate, &fc_isnull);
	if (fc_isnull)
		elog(ERROR, "unexpected null in pg_database");
	fc_version = get_collation_actual_version(fc_datlocprovider, TextDatumGetCString(fc_datum));

	ReleaseSysCache(fc_tp);

	if (fc_version)
		PG_RETURN_TEXT_P(cstring_to_text(fc_version));
	else
		PG_RETURN_NULL();
}


/*
 * 辅助函数
 */

/*
 * 查找名为 "name" 的数据库的信息。如果数据库存在，则获取其上指定的锁类型，填写任何非 NULL 的剩余参数，并返回 true。如果没有这样的数据库，则返回 false。
 */
static bool fc_get_db_info(const char *fc_name, LOCKMODE fc_lockmode,
			Oid *fc_dbIdP, Oid *fc_ownerIdP,
			int *fc_encodingP, bool *fc_dbIsTemplateP, bool *fc_dbAllowConnP,
			TransactionId *fc_dbFrozenXidP, MultiXactId *fc_dbMinMultiP,
			Oid *fc_dbTablespace, char **fc_dbCollate, char **fc_dbCtype, char **fc_dbIculocale,
			char *fc_dbLocProvider,
			char **fc_dbCollversion)
{
	bool		fc_result = false;
	Relation	fc_relation;

	AssertArg(fc_name);

	/* 调用者可能希望提前在 pg_database 上获取更好的锁... */
	fc_relation = table_open(DatabaseRelationId, AccessShareLock);

	/*
	 * 循环覆盖了在我们能够锁定它之前数据库被重命名的少见情况。我们再尝试一次，以防我们能找到一个同名的新数据库。
	 */
	for (;;)
	{
		ScanKeyData fc_scanKey;
		SysScanDesc fc_scan;
		HeapTuple	fc_tuple;
		Oid			fc_dbOid;

		/*
		 * 数据库索引按名称没有系统缓存，因此必须以麻烦的方式进行
		 */
		ScanKeyInit(&fc_scanKey,
					Anum_pg_database_datname,
					BTEqualStrategyNumber, F_NAMEEQ,
					CStringGetDatum(fc_name));

		fc_scan = systable_beginscan(fc_relation, DatabaseNameIndexId, true,
								  NULL, 1, &fc_scanKey);

		fc_tuple = systable_getnext(fc_scan);

		if (!HeapTupleIsValid(fc_tuple))
		{
			/* 肯定没有这样名称的数据库 */
			systable_endscan(fc_scan);
			break;
		}

		fc_dbOid = ((Form_pg_database) GETSTRUCT(fc_tuple))->oid;

		systable_endscan(fc_scan);

		/*
		 * 现在我们有了数据库 OID，我们可以尝试锁定数据库。
		 */
		if (fc_lockmode != NoLock)
			LockSharedObject(DatabaseRelationId, fc_dbOid, 0, fc_lockmode);

		/*
		 * 现在，通过 OID 重新获取元组。如果它仍然存在并且名称仍然相同，我们就成功了；否则，释放锁并循环回去再试一次。
		 */
		fc_tuple = SearchSysCache1(DATABASEOID, ObjectIdGetDatum(fc_dbOid));
		if (HeapTupleIsValid(fc_tuple))
		{
			Form_pg_database fc_dbform = (Form_pg_database) GETSTRUCT(fc_tuple);

			if (strcmp(fc_name, NameStr(fc_dbform->datname)) == 0)
			{
				Datum		fc_datum;
				bool		fc_isnull;

				/* 数据库的 oid */
				if (fc_dbIdP)
					*fc_dbIdP = fc_dbOid;
				/* 所有者的 oid */
				if (fc_ownerIdP)
					*fc_ownerIdP = fc_dbform->datdba;
				/* 字符编码 */
				if (fc_encodingP)
					*fc_encodingP = fc_dbform->encoding;
				/* 允许作为模板？ */
				if (fc_dbIsTemplateP)
					*fc_dbIsTemplateP = fc_dbform->datistemplate;
				/* 允许连接？ */
				if (fc_dbAllowConnP)
					*fc_dbAllowConnP = fc_dbform->datallowconn;
				/* 冻结的 XID 限制 */
				if (fc_dbFrozenXidP)
					*fc_dbFrozenXidP = fc_dbform->datfrozenxid;
				/* 最小 MultiXactId */
				if (fc_dbMinMultiP)
					*fc_dbMinMultiP = fc_dbform->datminmxid;
				/* 此数据库的默认表空间 */
				if (fc_dbTablespace)
					*fc_dbTablespace = fc_dbform->dattablespace;
				/* 此数据库的默认区域设置 */
				if (fc_dbLocProvider)
					*fc_dbLocProvider = fc_dbform->datlocprovider;
				if (fc_dbCollate)
				{
					fc_datum = SysCacheGetAttr(DATABASEOID, fc_tuple, Anum_pg_database_datcollate, &fc_isnull);
					Assert(!fc_isnull);
					*fc_dbCollate = TextDatumGetCString(fc_datum);
				}
				if (fc_dbCtype)
				{
					fc_datum = SysCacheGetAttr(DATABASEOID, fc_tuple, Anum_pg_database_datctype, &fc_isnull);
					Assert(!fc_isnull);
					*fc_dbCtype = TextDatumGetCString(fc_datum);
				}
				if (fc_dbIculocale)
				{
					fc_datum = SysCacheGetAttr(DATABASEOID, fc_tuple, Anum_pg_database_daticulocale, &fc_isnull);
					if (fc_isnull)
						*fc_dbIculocale = NULL;
					else
						*fc_dbIculocale = TextDatumGetCString(fc_datum);
				}
				if (fc_dbCollversion)
				{
					fc_datum = SysCacheGetAttr(DATABASEOID, fc_tuple, Anum_pg_database_datcollversion, &fc_isnull);
					if (fc_isnull)
						*fc_dbCollversion = NULL;
					else
						*fc_dbCollversion = TextDatumGetCString(fc_datum);
				}
				ReleaseSysCache(fc_tuple);
				fc_result = true;
				break;
			}
			/* 只有在刚重命名时才能到达这里 */
			ReleaseSysCache(fc_tuple);
		}

		if (fc_lockmode != NoLock)
			UnlockSharedObject(DatabaseRelationId, fc_dbOid, 0, fc_lockmode);
	}

	table_close(fc_relation, AccessShareLock);

	return fc_result;
}

/* 检查当前用户是否具有创建数据库的权限 */
static bool fc_have_createdb_privilege(void)
{
	bool		fc_result = false;
	HeapTuple	fc_utup;

	/* 超级用户始终可以做任何事情 */
	if (superuser())
		return true;

	fc_utup = SearchSysCache1(AUTHOID, ObjectIdGetDatum(GetUserId()));
	if (HeapTupleIsValid(fc_utup))
	{
		fc_result = ((Form_pg_authid) GETSTRUCT(fc_utup))->rolcreatedb;
		ReleaseSysCache(fc_utup);
	}
	return fc_result;
}

/*
 * 删除表空间目录
 *
 * 我们不知道 db_id 使用了哪些表空间，因此遍历所有表空间删除 <tablespace>/db_id
 */
static void fc_remove_dbtablespaces(Oid fc_db_id)
{
	Relation	fc_rel;
	TableScanDesc fc_scan;
	HeapTuple	fc_tuple;
	List	   *fc_ltblspc = NIL;
	ListCell   *fc_cell;
	int			fc_ntblspc;
	int			fc_i;
	Oid		   *fc_tablespace_ids;

	fc_rel = table_open(TableSpaceRelationId, AccessShareLock);
	fc_scan = table_beginscan_catalog(fc_rel, 0, NULL);
	while ((fc_tuple = heap_getnext(fc_scan, ForwardScanDirection)) != NULL)
	{
		Form_pg_tablespace fc_spcform = (Form_pg_tablespace) GETSTRUCT(fc_tuple);
		Oid			fc_dsttablespace = fc_spcform->oid;
		char	   *fc_dstpath;
		struct stat fc_st;

		/* 不要干扰全局表空间 */
		if (fc_dsttablespace == GLOBALTABLESPACE_OID)
			continue;

		fc_dstpath = GetDatabasePath(fc_db_id, fc_dsttablespace);

		if (lstat(fc_dstpath, &fc_st) < 0 || !S_ISDIR(fc_st.st_mode))
		{
			/* 假设我们可以忽略它 */
			pfree(fc_dstpath);
			continue;
		}

		if (!rmtree(fc_dstpath, true))
			ereport(WARNING,
					(errmsg("some useless files may be left behind in old database directory \"%s\"",
							fc_dstpath)));

		fc_ltblspc = lappend_oid(fc_ltblspc, fc_dsttablespace);
		pfree(fc_dstpath);
	}

	fc_ntblspc = list_length(fc_ltblspc);
	if (fc_ntblspc == 0)
	{
		table_endscan(fc_scan);
		table_close(fc_rel, AccessShareLock);
		return;
	}

	fc_tablespace_ids = (Oid *) palloc(fc_ntblspc * sizeof(Oid));
	fc_i = 0;
	foreach(fc_cell, fc_ltblspc)
		fc_tablespace_ids[fc_i++] = lfirst_oid(fc_cell);

	/* 在 XLOG 中记录文件系统更改 */
	{
		xl_dbase_drop_rec fc_xlrec;

		fc_xlrec.db_id = fc_db_id;
		fc_xlrec.ntablespaces = fc_ntblspc;

		XLogBeginInsert();
		XLogRegisterData((char *) &fc_xlrec, MinSizeOfDbaseDropRec);
		XLogRegisterData((char *) fc_tablespace_ids, fc_ntblspc * sizeof(Oid));

		(void) XLogInsert(RM_DBASE_ID,
						  XLOG_DBASE_DROP | XLR_SPECIAL_REL_UPDATE);
	}

	list_free(fc_ltblspc);
	pfree(fc_tablespace_ids);

	table_endscan(fc_scan);
	table_close(fc_rel, AccessShareLock);
}

/*
 * 检查是否存在与提议的新数据库 OID 冲突的现有文件；
 * 如果有任何文件，则返回 true
 *
 * 如果在任何表空间中有子目录匹配提议的新 OID，我们将因重复名称而导致创建失败……然后我们将在 remove_dbtablespaces 的清理过程中尝试删除那个已经存在的子目录。删除现有文件似乎是个坏主意，因此我们在确定新数据库的 OID 之前进行这个额外检查。这恰好与 GetNewRelFileNode() 对表 relfilenode 值所做的相对应。
 */
static bool fc_check_db_file_conflict(Oid fc_db_id)
{
	bool		fc_result = false;
	Relation	fc_rel;
	TableScanDesc fc_scan;
	HeapTuple	fc_tuple;

	fc_rel = table_open(TableSpaceRelationId, AccessShareLock);
	fc_scan = table_beginscan_catalog(fc_rel, 0, NULL);
	while ((fc_tuple = heap_getnext(fc_scan, ForwardScanDirection)) != NULL)
	{
		Form_pg_tablespace fc_spcform = (Form_pg_tablespace) GETSTRUCT(fc_tuple);
		Oid			fc_dsttablespace = fc_spcform->oid;
		char	   *fc_dstpath;
		struct stat fc_st;

		/* 不要干扰全局表空间 */
		if (fc_dsttablespace == GLOBALTABLESPACE_OID)
			continue;

		fc_dstpath = GetDatabasePath(fc_db_id, fc_dsttablespace);

		if (lstat(fc_dstpath, &fc_st) == 0)
		{
			/* 找到一个冲突的文件（或目录，无论如何） */
			pfree(fc_dstpath);
			fc_result = true;
			break;
		}

		pfree(fc_dstpath);
	}

	table_endscan(fc_scan);
	table_close(fc_rel, AccessShareLock);

	return fc_result;
}

/*
 * 为忙碌的数据库发出合适的 errdetail 消息
 */
static int fc_errdetail_busy_db(int fc_notherbackends, int fc_npreparedxacts)
{
	if (fc_notherbackends > 0 && fc_npreparedxacts > 0)

		/*
		 * 这里我们不处理单数与复数，因为 gettext
		 * 不支持一个字符串中的多个复数形式。
		 */
		errdetail("There are %d other session(s) and %d prepared transaction(s) using the database.",
				  fc_notherbackends, fc_npreparedxacts);
	else if (fc_notherbackends > 0)
		errdetail_plural("There is %d other session using the database.",
						 "There are %d other sessions using the database.",
						 fc_notherbackends,
						 fc_notherbackends);
	else
		errdetail_plural("There is %d prepared transaction using the database.",
						 "There are %d prepared transactions using the database.",
						 fc_npreparedxacts,
						 fc_npreparedxacts);
	return 0;					/* 仅用于保持 ereport 宏正常 */
}

/*
 * get_database_oid - 给定数据库名称，查找 OID
 *
 * 如果 missing_ok 为 false，则在未找到数据库名称时抛出错误。如果
 * 为 true，则只是返回 InvalidOid。
 */
Oid get_database_oid(const char *fc_dbname, bool fc_missing_ok)
{
	Relation	fc_pg_database;
	ScanKeyData fc_entry[1];
	SysScanDesc fc_scan;
	HeapTuple	fc_dbtuple;
	Oid			fc_oid;

	/*
	 * pg_database 中没有按名称索引的 syscache，因此我们必须通过
	 * 硬方式查找。
	 */
	fc_pg_database = table_open(DatabaseRelationId, AccessShareLock);
	ScanKeyInit(&fc_entry[0],
				Anum_pg_database_datname,
				BTEqualStrategyNumber, F_NAMEEQ,
				CStringGetDatum(fc_dbname));
	fc_scan = systable_beginscan(fc_pg_database, DatabaseNameIndexId, true,
							  NULL, 1, fc_entry);

	fc_dbtuple = systable_getnext(fc_scan);

	/* 我们假设最多只有一个匹配的元组 */
	if (HeapTupleIsValid(fc_dbtuple))
		fc_oid = ((Form_pg_database) GETSTRUCT(fc_dbtuple))->oid;
	else
		fc_oid = InvalidOid;

	systable_endscan(fc_scan);
	table_close(fc_pg_database, AccessShareLock);

	if (!OidIsValid(fc_oid) && !fc_missing_ok)
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_DATABASE),
				 errmsg("database \"%s\" does not exist",
						fc_dbname)));

	return fc_oid;
}


/*
 * get_database_name - 给定数据库 OID，查找名称
 *
 * 返回一个 palloc'd 字符串，如果没有此数据库，则返回 NULL。
 */
char * get_database_name(Oid fc_dbid)
{
	HeapTuple	fc_dbtuple;
	char	   *fc_result;

	fc_dbtuple = SearchSysCache1(DATABASEOID, ObjectIdGetDatum(fc_dbid));
	if (HeapTupleIsValid(fc_dbtuple))
	{
		fc_result = pstrdup(NameStr(((Form_pg_database) GETSTRUCT(fc_dbtuple))->datname));
		ReleaseSysCache(fc_dbtuple);
	}
	else
		fc_result = NULL;

	return fc_result;
}


/*
 * 在删除数据库时，pg_database 行被标记为无效，但目录内容仍然存在。
 * 不允许连接到这样的数据库。
 */
bool database_is_invalid_form(Form_pg_database fc_datform)
{
	return fc_datform->datconnlimit == DATCONNLIMIT_INVALID_DB;
}


/*
 * database_is_invalid_form() 的便捷包装
 */
bool database_is_invalid_oid(Oid fc_dboid)
{
	HeapTuple	fc_dbtup;
	Form_pg_database fc_dbform;
	bool		fc_invalid;

	fc_dbtup = SearchSysCache1(DATABASEOID, ObjectIdGetDatum(fc_dboid));
	if (!HeapTupleIsValid(fc_dbtup))
		elog(ERROR, "cache lookup failed for database %u", fc_dboid);
	fc_dbform = (Form_pg_database) GETSTRUCT(fc_dbtup);

	fc_invalid = database_is_invalid_form(fc_dbform);

	ReleaseSysCache(fc_dbtup);

	return fc_invalid;
}


/*
 * recovery_create_dbdir()
 *
 * 在恢复期间，有一种情况我们需要有效地恢复缺失的
 * 表空间目录，以便恢复可以继续。这发生在
 * 恢复想要创建一个数据库，但持有的表空间在服务器停止之前被删除。
 * 由于我们预计在达到恢复一致性之前目录就会消失，并且在这里除了
 * OID 外我们对表空间没有任何了解，因此我们在
 * pg_tblspc 下创建一个真实的目录，而不是恢复符号链接。
 *
 * 如果 only_tblspc 为 true，则请求的目录必须位于 pg_tblspc/
 */
static void fc_recovery_create_dbdir(char *fc_path, bool fc_only_tblspc)
{
	struct stat fc_st;

	Assert(RecoveryInProgress());

	if (stat(fc_path, &fc_st) == 0)
		return;

	if (fc_only_tblspc && strstr(fc_path, "pg_tblspc/") == NULL)
		elog(PANIC, "requested to created invalid directory: %s", fc_path);

	if (reachedConsistency && !allow_in_place_tablespaces)
		ereport(PANIC,
				errmsg("missing directory \"%s\"", fc_path));

	elog(reachedConsistency ? WARNING : DEBUG1,
		 "creating missing directory: %s", fc_path);

	if (pg_mkdir_p(fc_path, pg_dir_create_mode) != 0)
		ereport(PANIC,
				errmsg("could not create missing directory \"%s\": %m", fc_path));
}


/*
 * DATABASE 资源管理器的例程
 */
void dbase_redo(XLogReaderState *fc_record)
{
	uint8		fc_info = XLogRecGetInfo(fc_record) & ~XLR_INFO_MASK;

	/* 在 dbase 记录中不使用备份块 */
	Assert(!XLogRecHasAnyBlockRefs(fc_record));

	if (fc_info == XLOG_DBASE_CREATE_FILE_COPY)
	{
		xl_dbase_create_file_copy_rec *fc_xlrec =
		(xl_dbase_create_file_copy_rec *) XLogRecGetData(fc_record);
		char	   *fc_src_path;
		char	   *fc_dst_path;
		char	   *fc_parent_path;
		struct stat fc_st;

		fc_src_path = GetDatabasePath(fc_xlrec->src_db_id, fc_xlrec->src_tablespace_id);
		fc_dst_path = GetDatabasePath(fc_xlrec->db_id, fc_xlrec->tablespace_id);

		/*
		 * 我们对重放 CREATE 的理论是强制删除目标
		 * 子目录（如果存在），然后重新复制源数据。这可能是
		 * 超出需要的工作，但实施起来很简单。
		 */
		if (stat(fc_dst_path, &fc_st) == 0 && S_ISDIR(fc_st.st_mode))
		{
			if (!rmtree(fc_dst_path, true))
				/* 如果这失败了，下面的 copydir() 将出错。 */
				ereport(WARNING,
						(errmsg("some useless files may be left behind in old database directory \"%s\"",
								fc_dst_path)));
		}

		/*
		 * 如果目标路径的父级不存在，现在就创建它。这
		 * 使我们能够稍后在其下方创建目标。请注意，如果
		 * 数据库目录不在表空间中，父级将始终
		 * 存在，因此在这种情况下这永远不会运行。
		 */
		fc_parent_path = pstrdup(fc_dst_path);
		get_parent_directory(fc_parent_path);
		if (stat(fc_parent_path, &fc_st) < 0)
		{
			if (errno != ENOENT)
				ereport(FATAL,
						errmsg("could not stat directory \"%s\": %m",
							   fc_dst_path));

			fc_recovery_create_dbdir(fc_parent_path, true);
		}
		pfree(fc_parent_path);

		/*
		 * 有一种情况是由于上述原因复制源目录丢失。
		 * 创建空源目录，以便下面的 copydir 不会失败。
		 * 该目录将在恢复时很快被删除。
		 */
		if (stat(fc_src_path, &fc_st) < 0 && errno == ENOENT)
			fc_recovery_create_dbdir(fc_src_path, false);

		/*
		 * 将脏缓冲区强制写出到磁盘，以确保源数据库
		 * 是最新的以进行复制。
		 */
		FlushDatabaseBuffers(fc_xlrec->src_db_id);

		/* 关闭所有后端中的所有 sgmr fds。 */
		WaitForProcSignalBarrier(EmitProcSignalBarrier(PROCSIGNAL_BARRIER_SMGRRELEASE));

		/*
		 * 将此子目录复制到新位置
		 *
		 * 我们不需要复制子目录
		 */
		copydir(fc_src_path, fc_dst_path, false);

		pfree(fc_src_path);
		pfree(fc_dst_path);
	}
	else if (fc_info == XLOG_DBASE_CREATE_WAL_LOG)
	{
		xl_dbase_create_wal_log_rec *fc_xlrec =
		(xl_dbase_create_wal_log_rec *) XLogRecGetData(fc_record);
		char	   *fc_dbpath;
		char	   *fc_parent_path;

		fc_dbpath = GetDatabasePath(fc_xlrec->db_id, fc_xlrec->tablespace_id);

		/* 如有需要并且有效，创建父目录 */
		fc_parent_path = pstrdup(fc_dbpath);
		get_parent_directory(fc_parent_path);
		fc_recovery_create_dbdir(fc_parent_path, true);

		/* 创建带有版本文件的数据库目录。 */
		fc_CreateDirAndVersionFile(fc_dbpath, fc_xlrec->db_id, fc_xlrec->tablespace_id,
								true);
		pfree(fc_dbpath);
	}
	else if (fc_info == XLOG_DBASE_DROP)
	{
		xl_dbase_drop_rec *fc_xlrec = (xl_dbase_drop_rec *) XLogRecGetData(fc_record);
		char	   *fc_dst_path;
		int			fc_i;

		if (InHotStandby)
		{
			/*
			 * 在解决冲突时锁定数据库，以确保
			 * InitPostgres()无法并发完全重新执行。这
			 * 避免了后端自动重新连接到同一数据库，
			 * 在某些情况下可能会发生这种情况。
			 *
			 * 这也将锁定尝试连接到数据库特定
			 * 插槽的walsenders，因此我们可以安全地删除
			 * 插槽。
			 */
			LockSharedObjectForSession(DatabaseRelationId, fc_xlrec->db_id, 0, AccessExclusiveLock);
			ResolveRecoveryConflictWithDatabase(fc_xlrec->db_id);
		}

		/* 删除任何数据库特定的复制插槽 */
		ReplicationSlotsDropDBSlots(fc_xlrec->db_id);

		/* 删除此数据库在共享缓冲区缓存中的页面 */
		DropDatabaseBuffers(fc_xlrec->db_id);

		/* 此外，还清除md.c中可能挂起的任何fsync请求 */
		ForgetDatabaseSyncRequests(fc_xlrec->db_id);

		/* 还清理xlog relcache */
		XLogDropDatabase(fc_xlrec->db_id);
		
#ifdef FDD
		/* 删除此数据库的任何逻辑故障转移插槽 */
		ReplicationSlotsDropDBSlots(fc_xlrec->db_id);
#endif

		/* 关闭所有后端中的所有 sgmr fds。 */
		WaitForProcSignalBarrier(EmitProcSignalBarrier(PROCSIGNAL_BARRIER_SMGRRELEASE));

		for (fc_i = 0; fc_i < fc_xlrec->ntablespaces; fc_i++)
		{
			fc_dst_path = GetDatabasePath(fc_xlrec->db_id, fc_xlrec->tablespace_ids[fc_i]);

			/* 并移除物理文件 */
			if (!rmtree(fc_dst_path, true))
				ereport(WARNING,
						(errmsg("some useless files may be left behind in old database directory \"%s\"",
								fc_dst_path)));
			pfree(fc_dst_path);
		}

		if (InHotStandby)
		{
			/*
			 * 在提交之前释放锁。XXX这里存在竞争条件
			 * 可能允许后端重新连接，但这个窗口很小，
			 * 因为这里和提交之间的间隔大多数
			 * 都比较小，而且不太可能有人会删除
			 * 我们试图连接的数据库。
			 */
			UnlockSharedObjectForSession(DatabaseRelationId, fc_xlrec->db_id, 0, AccessExclusiveLock);
		}
	}
	else
		elog(PANIC, "dbase_redo: unknown op code %u", fc_info);
}
